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.

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.

Không cần kết quả thì dùng execute, đừng dùng submit. Còn nếu dùng submit, phải thật sự gọi get() ở đâu đó. Một tác vụ nền chết lặng lẽ là loại lỗi tệ nhất: chức năng ngừng hoạt động, log sạch bong, và không ai biết bắt đầu tìm từ đâu.

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:

  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 — hàng đợi không chứa gì (SynchronousQueue), nhưng số luồng 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. 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à 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.

Thử ba mươi giây

Tìm trong dự án của bạn chuỗi Executors.new.

Mỗi chỗ tìm được, hỏi: 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.

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.