newFixedThreadPool và newCachedThreadPool: hai dòng code tiện tay là hai quả bom hẹn giờ

Hãy nghĩ về một tổng đài chăm sóc khách hàng. Nó có một số nhân viên cố định và một phòng chờ có sức chứa nhất định. Khi cả hai đầy, người quản lý phải quyết định: báo bận, để khách chờ thêm, hay tự mình nhấc máy. Thread pool trong Java là cái tổng đài đó — và hai cách dựng pool mà ai cũng gõ theo phản xạ lại là hai cách bỏ đi giới hạn: một cái cho phòng chờ dài vô tận, một cái thuê nhân viên vô hạn. Cả hai chạy mượt tới đúng ngày đông khách.

Bảy bài vừa qua đều tự tạo Thread hoặc dùng Executors.newFixedThreadPool cho gọn. Hôm nay là bài về chính công cụ đó — và về việc nó có nhiều cạm bẫy hơn vẻ ngoài rất nhiều. (Các con số dưới đây đo trên máy 16 nhân của tôi; tuyệt đối thì tuỳ máy, nhưng hình dạng của các đường mới là thứ đáng đọc.)

Bắt đầu bằng cái tôi nghĩ là nguy hiểm nhất, vì nó im lặng tuyệt đối.

submit() nuốt ngoại lệ

es.submit(() -> { throw new IllegalStateException("nổ trong tác vụ"); });

  -> không in gì ra cả, không log gì cả

Không có gì. Tác vụ chết, và bạn không biết.

Ngoại lệ được cất vào Future, và nó chỉ hiện ra khi có ai đó gọi get():

  f.get() -> java.lang.IllegalStateException: nổ trong tác vụ

Mà submit trả về Future thì rất nhiều người bỏ qua giá trị trả về — nhất là với tác vụ chạy nền, nơi chẳng có kết quả nào để lấy.

Cùng đoạn mã đó với execute:

Exception in thread "pool-1-thread-1" java.lang.IllegalStateException: nổ trong tác vụ
	at Exec.lambda$main$1(Exec.java:17)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(...)

In đầy đủ ra stderr.

Cách chắc chắn nhất là bọc thân tác vụ trong try/catch và tự ghi log. Đặt Thread.setDefaultUncaughtExceptionHandler cũng có ích, nhưng nó không bắt được ngoại lệ từ submit — vì với JVM, tác vụ đó không hề ném ngoại lệ ra ngoài; nó kết thúc bình thường và cất lỗi vào Future.

Một hệ quả nữa ít người biết: với execute, luồng bị chết sẽ được pool thay bằng luồng mới, nên pool không teo dần. Nhưng tên luồng sẽ nhảy số (pool-1-thread-1 rồi -7, -13…), và đó là dấu hiệu đáng ngờ khi đọc log.

Hai kiểu pool phổ biến nhất đều là bom hẹn giờ

Executors.newFixedThreadPool(n) dùng hàng đợi không giới hạn — phòng chờ dài vô tận:

  nộp 100.000 tác vụ vào pool 2 luồng -> hàng đợi giữ 99.998 tác vụ, không hề từ chối

Không một lời từ chối. Tác vụ đến nhanh hơn tốc độ xử lý thì hàng đợi cứ lớn lên tới khi hết bộ nhớ. Và vì mỗi tác vụ thường giữ tham chiếu tới dữ liệu của nó, thứ tích tụ không chỉ là vài con trỏ.

Đây đúng là cái bẫy LinkedBlockingQueue không giới hạn ở bài hôm qua, chỉ là được gói lại dưới một cái tên nghe rất vô hại.

Executors.newCachedThreadPool() thì ngược lại — phòng chờ không chứa gì (SynchronousQueue), nhưng số nhân viên không giới hạn:

  nộp 3.000 tác vụ -> pool tạo 3.000 luồng nền tảng

Ba nghìn luồng của hệ điều hành. Bài 67 đã đo: khoảng 35.000 luồng là máy đứng hẳn, không kèm ngoại lệ nào.

Cả hai kiểu đều không có áp lực ngược. Chúng nhận mọi thứ bạn ném vào, và trả giá ở một chỗ khác vào một lúc khác.

Tự dựng, và chọn chuyện gì xảy ra khi đầy

new ThreadPoolExecutor(
        soLuongLoi,                        // luồng giữ thường trực
        soLuongToiDa,                      // trần
        60, TimeUnit.SECONDS,              // luồng dư nhàn rỗi bao lâu thì bỏ
        new ArrayBlockingQueue<>(1000),    // hàng đợi CÓ GIỚI HẠN
        new ThreadPoolExecutor.CallerRunsPolicy());

Thứ tự hoạt động của nó không hiển nhiên, và đáng thuộc: pool tạo luồng tới soLuongLoi, rồi xếp hàng đợi cho đầy, rồi mới tạo thêm luồng tới soLuongToiDa, rồi mới từ chối.

Nghĩa là với hàng đợi lớn, soLuongToiDa gần như không bao giờ được dùng tới — một chi tiết khiến nhiều cấu hình không hoạt động như người viết tưởng.

Bốn chính sách khi đã đầy cả hàng lẫn luồng, đo với pool 1 luồng, hàng đợi 2, nộp 10 tác vụ:

  AbortPolicy         : chạy  3, ném lỗi 7, chạy trên chính luồng gọi 0
  CallerRunsPolicy    : chạy 10, ném lỗi 0, chạy trên chính luồng gọi 4
  DiscardPolicy       : chạy  3, ném lỗi 0, chạy trên chính luồng gọi 0
  DiscardOldestPolicy : chạy  3, ném lỗi 0, chạy trên chính luồng gọi 0

AbortPolicy (mặc định) ném RejectedExecutionException — báo bận thẳng thừng. Trung thực: bạn biết ngay là quá tải.

CallerRunsPolicy cho chạy trên chính luồng gọi — cột cuối cho thấy 4 trong 10 tác vụ chạy như vậy. Đây là người quản lý tự nhấc máy, và là lựa chọn tôi thích nhất cho máy chủ: luồng nộp việc bị bận nên tự nó chậm lại, tạo ra áp lực ngược tự nhiên mà không mất tác vụ nào. Đánh đổi là luồng gọi bị chặn — với luồng xử lý request thì đó chính là điều bạn muốn; với luồng điều phối duy nhất thì đó là thảm hoạ.

DiscardPolicy vứt lặng lẽ. DiscardOldestPolicy vứt cái chờ lâu nhất rồi thử lại. Cả hai chỉ đúng khi mất dữ liệu là chấp nhận được — số liệu giám sát, cập nhật vị trí theo thời gian thực. Đừng dùng cho thứ gì người dùng trông đợi.

Kích thước pool: hai loại tải, hai đáp án ngược nhau

Máy 16 nhân, 200 tác vụ.

Tải tốn CPU:

  pool=  1 ->    8 ms
  pool=  8 ->    2 ms
  pool= 16 ->    2 ms
  pool= 32 ->    4 ms
  pool= 64 ->    7 ms
  pool=128 ->   18 ms

Tốt nhất quanh số nhân, rồi xấu dần. Pool 128 chậm hơn pool 16 chín lần — thêm luồng không thêm CPU, chỉ thêm việc chuyển ngữ cảnh.

Tải chờ I/O (mỗi tác vụ ngủ 50 ms):

  pool=  1 -> 10260 ms
  pool= 16 ->   666 ms
  pool= 32 ->   357 ms
  pool= 64 ->   206 ms
  pool=128 ->   109 ms
  pool=200 ->    73 ms

Càng lớn càng nhanh, tới tận bằng số tác vụ. Ở đây "pool bằng số nhân" là lời khuyên tệ hại: nó chậm hơn chín lần so với pool 200.

Đây đúng là kết luận của bài 66 về pool kết nối, gặp lại ở một tầng khác: kích thước phải bằng chỗ nghẽn thật, không bằng một công thức chung.

Công thức khởi đầu quen dùng:

số luồng ≈ số nhân × (1 + thời gian chờ / thời gian tính)

Tác vụ thuần tính toán thì tỷ số bằng 0, ra đúng số nhân. Tác vụ chờ mạng 50 ms và tính 1 ms thì ra khoảng 50 lần số nhân. Nhưng như mọi công thức, nó là điểm bắt đầu để đo, không phải đáp án.

Hoặc thôi không chọn nữa

  newVirtualThreadPerTaskExecutor -> 59 ms  (không phải chọn kích thước)

Nhanh hơn cả pool 200 luồng nền tảng, và không có tham số nào để đặt sai.

try (var es = Executors.newVirtualThreadPerTaskExecutor()) {
    for (var viec : danhSach) es.submit(viec);
}   // close() tự chờ mọi tác vụ xong

Chú ý cái try-với-tài-nguyên: từ Java 21, ExecutorService cài AutoCloseable, và close() gọi shutdown() rồi chờ. Không còn phải viết shutdown() + awaitTermination() bằng tay.

Với tải chờ I/O, đây gần như luôn là lựa chọn đúng trên Java 21. Với tải tốn CPU thì không — mỗi luồng ảo vẫn cần một luồng mang, nên tạo mười nghìn luồng ảo tính toán không nhanh hơn pool bằng số nhân, chỉ tốn thêm bộ nhớ. Với tải CPU, newFixedThreadPool(số nhân) vẫn đúng.

Và nhớ cảnh báo ở bài 68: luồng ảo chặn trong synchronized thì bị ghim, và lợi ích bốc hơi.

Tắt cho đúng

  shutdown()   : tác vụ hoàn thành 10/10 | trả về 0 tác vụ chưa chạy
  shutdownNow(): tác vụ hoàn thành  0/10 | trả về 8 tác vụ chưa chạy

shutdown() từ chối tác vụ mới nhưng chạy nốt mọi thứ đã nhận, kể cả những cái còn trong hàng đợi.

shutdownNow() ngắt các luồng đang chạy và trả về danh sách tác vụ chưa kịp chạy. Trong phép đo, không tác vụ nào hoàn thành — kể cả hai cái đang chạy dở, vì chúng bị interrupt() giữa chừng.

Hai điều cần nhớ:

Cả hai đều không chờ. Chúng trả về ngay. Muốn chờ thật thì phải awaitTermination.

shutdownNow chỉ ngắt được tác vụ biết nghe. Đúng bài học ở bài 67: một vòng lặp không kiểm cờ ngắt, hoặc một khối catch (InterruptedException e) {} trống, sẽ chạy tiếp bất chấp.

Khuôn tắt máy đầy đủ, đáng chép sẵn:

es.shutdown();
try {
    if (!es.awaitTermination(30, TimeUnit.SECONDS)) {
        es.shutdownNow();
        if (!es.awaitTermination(10, TimeUnit.SECONDS))
            log.error("pool không chịu tắt");
    }
} catch (InterruptedException e) {
    es.shutdownNow();
    Thread.currentThread().interrupt();
}

invokeAll: nộp cả loạt rồi chờ

List<Future<Long>> kq = es.invokeAll(danhSachCallable);

  invokeAll 8 tác vụ -> 8 kết quả (mọi Future đã xong)
  isDone của phần tử đầu: true

invokeAll chặn cho tới khi tất cả xong, nên mọi Future trả về đều đã hoàn thành — không phải kiểm tra gì thêm. Nó tránh luôn cái bẫy stream lười ở bài 64, vì việc gửi đi và việc chờ tách bạch rõ ràng.

Có bản nhận hạn chờ, và có invokeAny trả về kết quả của cái xong đầu tiên rồi huỷ phần còn lại — hợp khi hỏi nhiều nguồn và chỉ cần một câu trả lời.

Danh sách kiểm khi dựng pool

Đừng dùng Executors.newFixedThreadPool và newCachedThreadPool cho mã sản xuất. Dựng ThreadPoolExecutor với hàng đợi có giới hạn.

Đặt tên luồng. Truyền một ThreadFactory đặt tên theo mục đích. pool-3-thread-7 không nói gì trong ảnh chụp ngăn xếp; xu-ly-anh-7 thì có.

Chọn chính sách từ chối có chủ đích, đừng để mặc định rồi ngạc nhiên khi gặp RejectedExecutionException.

Đừng chờ kết quả của tác vụ nộp vào cùng pool. Đó là biến thể deadlock ở bài 72, và jstack sẽ không gọi tên nó.

Theo dõi getQueue().size() và getActiveCount(). Hàng đợi dài dần là dấu hiệu quá tải, hiện ra trước khi người dùng kịp phàn nàn.

Muốn tự soi trong ba mươi giây: tìm trong dự án chuỗi Executors.new, và với mỗi chỗ, hỏi một câu — nếu tác vụ đến nhanh gấp mười lần tốc độ xử lý trong năm phút, chuyện gì xảy ra? Với newFixedThreadPool thì hàng đợi phình ra; với newCachedThreadPool thì số luồng phình ra. Cả hai đều chạy hoàn hảo cho tới đúng ngày lưu lượng tăng gấp mười.

Mẫu số chung

Cái tổng đài nhân-viên-cố-định-cộng-phòng-chờ không phải phát minh của Java — nó là hình mẫu của mọi hệ thống có số thợ hữu hạn nhận việc đến vô hạn, nên bài học áp được ở mọi ngôn ngữ. Go không phát cho bạn cái bẫy "hàng đợi không giới hạn", nhưng bạn tự dựng áp lực ngược bằng một channel có đệm làm semaphore, hoặc errgroup.SetLimit(n) — chính là chọn số thợ. Python có concurrent.futures với ThreadPoolExecutor và ProcessPoolExecutor, cùng submit/Future y hệt Java, và bên asyncio thì dùng Semaphore để chặn số việc chạy song song. Node có sẵn một thread pool của libuv với UV_THREADPOOL_SIZE mặc định chỉ 4 — một pool cố định bạn buộc phải biết mà chỉnh, nếu không mọi thao tác tệp và mã hoá đều nghẽn qua đúng bốn luồng đó. Và pool kết nối cơ sở dữ liệu (HikariCP) là đúng cái tổng đài này đội lốt khác, với đúng bài học "kích thước bằng chỗ nghẽn thật".

Điểm chung, và là thứ đáng mang theo, gồm hai vế. Vế thứ nhất: bất kỳ hàng đợi công việc nào không có giới hạn đều là một quả bom hẹn giờ — thiếu áp lực ngược thì tải cao không bị từ chối, nó chỉ dời sang chỗ khác (bộ nhớ, số luồng) và nổ muộn hơn, xa điểm gây ra hơn, khó lần hơn. Cách chữa ở đâu cũng là đặt một cái trần rồi quyết định tỉnh táo chuyện gì xảy ra khi chạm trần. Vế thứ hai: kích thước pool cho tải CPU và cho tải I/O ngược nhau — CPU thì quanh số nhân rồi tệ dần, I/O thì càng lớn càng nhanh tới tận số việc — nên không có "số luồng đúng" chung, chỉ có số luồng bằng đúng chỗ nghẽn mà bạn phải đo để biết.

Ngày mai: CompletableFuture — ghép nối các bước bất đồng bộ, xử lý lỗi trong chuỗi, và chỗ nó khác Future thường.