Future của Java 5 có một vấn đề: cách duy nhất để lấy kết quả là get(), mà get() thì chặn. Bạn không ghép nối được các bước, không xử lý lỗi trong chuỗi, không gộp hai kết quả.

CompletableFuture (Java 8) giải quyết cả ba. Đổi lại, nó có một cái bẫy mà tôi thấy gây sự cố sản xuất nhiều hơn mọi thứ khác trong bài — nên ta bắt đầu từ đó.

Pool mặc định nhỏ hơn bạn tưởng

  số nhân: 4 | pool chung có 3 luồng

supplyAsync không có executor thì chạy trên ForkJoinPool.commonPool(), và pool ấy có số nhân trừ một luồng. Trên máy bốn nhân là ba.

Mười tác vụ, mỗi tác vụ chờ 500 mili giây:

  dùng pool chung           : 2004 ms
  dùng executor riêng 10 luồng :  503 ms

Bốn lần. Mười tác vụ chờ I/O đáng lẽ xong trong 500 ms lại mất hai giây, vì chúng phải chia nhau ba luồng.

Nhưng đó chưa phải phần tệ nhất. Pool chung là tài nguyên dùng chung cho cả JVM — parallelStream dùng nó, mọi thư viện trong classpath dùng nó, mọi phần khác của ứng dụng bạn cũng dùng nó.

Tôi nộp 24 tác vụ chặn 4 giây vào pool chung rồi đo phần còn lại:

  getPoolSize()=3  getActiveThreadCount()=3  getQueuedSubmissionCount()=21

  parallelStream  : 58 ms khi rảnh  ->  189 ms lúc này
  một supplyAsync đơn lẻ           : 31 344 ms

Ba mươi mốt giây cho một tác vụ tính toán tầm thường. Nó không chậm — nó xếp hàng sau 21 tác vụ đang chờ.

Chú ý parallelStream chỉ chậm 3,3 lần chứ không đứng hẳn: luồng gọi parallelStream cũng tham gia làm việc, nên nó vẫn tiến được. Còn supplyAsync thì không có ai làm hộ — nó chỉ biết chờ tới lượt.

Đừng bao giờ chặn trong pool chung. Mọi lời gọi mạng, truy vấn cơ sở dữ liệu, đọc tệp hay Thread.sleep trong supplyAsync không có executor đều là đang giam một trong số ít ỏi những luồng mà cả JVM dùng chung. Luôn truyền executor riêng: supplyAsync(viec, executorCuaToi).

Ba con số getPoolSize, getActiveThreadCount, getQueuedSubmissionCount ở trên rất đáng đưa lên bảng giám sát — getQueuedSubmissionCount lớn dần là dấu hiệu ai đó đang chặn pool chung.

thenApply chạy trên luồng nào

Câu trả lời là "còn tuỳ", và đó là điều bất ngờ đầu tiên:

  supplyAsync : ForkJoinPool.commonPool-worker-1
  thenApply   : main
  thenApply trên future ĐÃ xong: main   <- chạy ngay trên luồng gọi!

Quy tắc: thenApply chạy trên luồng nào hoàn thành bước trước. Nếu bước trước đã xong từ lúc bạn gắn thenApply vào, nó chạy ngay lập tức trên luồng đang gọi, đồng bộ.

Nghĩa là thenApply không đảm bảo bất đồng bộ. Đặt một việc nặng vào đó là bạn có thể vừa chặn luồng xử lý request mà không hề biết.

Muốn chắc chắn chạy trên luồng khác thì dùng bản Async kèm executor:

.thenApplyAsync(v -> viecNang(v), executorCuaToi)

Đây cũng là lý do các bản Async không kèm executor lại nguy hiểm gấp đôi: chúng vừa bất đồng bộ, vừa mặc định dùng pool chung.

Ba cách nối bước, và cách chọn

thenApply — biến đổi giá trị. Hàm trả về một giá trị thường.

thenCompose — nối sang một tác vụ bất đồng bộ khác. Hàm trả về một CompletableFuture.

  thenApply   -> CompletableFuture   (Future lồng trong Future)
  thenCompose -> Tên của user-7

Dùng nhầm thì nhận về CompletableFuture<CompletableFuture<String>> — biên dịch được, chạy được, và giá trị bạn cầm là một future chứ không phải kết quả. Nếu bạn quen Optional hay Stream: thenApply là map, thenCompose là flatMap, cùng một câu chuyện.

thenCombine — gộp hai chuỗi độc lập:

  hồ sơ + đơn hàng trong 303 ms   (hai lời gọi 300 ms chạy song song)

Hai lời gọi 300 ms xong trong 303 ms. Đây là chỗ CompletableFuture đáng giá nhất: gọi ba dịch vụ độc lập rồi gộp kết quả, thay vì cộng dồn thời gian.

Ngoại lệ đi thẳng qua các bước giữa

CompletableFuture.<Integer>supplyAsync(() -> { throw new IllegalStateException("nguồn hỏng"); })
        .thenApply(v -> { System.out.println("bước 1 CHẠY"); return v*2; })
        .thenApply(v -> { System.out.println("bước 2 CHẠY"); return v+1; })
        .exceptionally(e -> { ... });

  -> exceptionally bắt được: nguồn hỏng
  (hai bước giữa không hề chạy)

Một bước hỏng thì mọi bước sau bị bỏ qua cho tới chỗ xử lý lỗi đầu tiên. Giống hệt try/catch bao quanh một chuỗi lời gọi, chỉ khác là nó hoạt động qua nhiều luồng và nhiều thời điểm.

Ba cách xử lý, khác nhau ở chỗ tinh tế:

  exceptionally : thay thế                    <- chỉ chạy khi có lỗi, trả giá trị thay thế
  handle        : lỗi: hỏng rồi               <- luôn chạy, nhận cả (giá trị, lỗi)
  whenComplete  : quan sát được lỗi, KHÔNG nuốt nó
                  -> lỗi vẫn ném ra: hỏng rồi

exceptionally như catch. handle như catch cộng finally có giá trị trả về. whenComplete như finally thuần: nó quan sát rồi để lỗi đi tiếp — rất hợp để ghi log hoặc dọn tài nguyên mà không thay đổi kết quả.

Chọn sai giữa handle và whenComplete là lỗi rất dễ mắc: handle nuốt lỗi (chuỗi tiếp tục với giá trị bạn trả về), whenComplete thì không.

Cái bẫy im lặng, lần nữa

  đã tạo một future hỏng và không gọi join/get -> không log gì cả

Đúng vấn đề của submit() ở bài hôm qua, ở một hình dạng khác. Chuỗi CompletableFuture mà không ai gọi join(), get() hay gắn exceptionally sẽ thất bại trong im lặng tuyệt đối.

Với chuỗi chạy nền và không có ai chờ kết quả, hãy luôn kết thúc bằng một chỗ ghi log:

.whenComplete((v, e) -> { if (e != null) log.error("chuỗi xử lý hỏng", e); });

Và về join() với get():

  join() ném : CompletionException  (không kiểm tra)
  get()  ném : ExecutionException   (phải khai throws)

Cả hai đều bọc ngoại lệ gốc, nên nhớ e.getCause(). Tôi dùng join() trong lambda và stream vì nó không bắt khai throws; get() khi cần hạn chờ.

Hạn chờ, có từ Java 9

  completeOnTimeout -> 'giá trị dự phòng' sau 307 ms
  orTimeout         -> TimeoutException  sau 303 ms

orTimeout làm chuỗi hỏng khi quá hạn; completeOnTimeout thay bằng một giá trị dự phòng.

Cả hai đáng có ở mọi chỗ gọi dịch vụ ngoài, đúng lý do đã nói ở bài 64: một dịch vụ chậm mà không có hạn chờ sẽ ăn dần tài nguyên của bạn.

Một điều cần biết: hạn chờ không huỷ công việc đang chạy. Tác vụ vẫn tiếp tục ngốn luồng cho tới khi nó tự xong — bạn chỉ thôi chờ nó thôi. CompletableFuture không có cách nào ngắt một tác vụ đang chạy, vì nó không biết tác vụ đang chạy trên luồng nào.

allOf và anyOf

  anyOf -> B sau 101 ms          (cái nhanh nhất)
  allOf -> [A, B, C] sau 191 ms  (chờ cái chậm nhất)

allOf trả về CompletableFuture<Void> — nó chỉ báo "xong hết rồi", không gom kết quả. Lấy kết quả thì join() từng cái sau đó, và lúc ấy chúng đã xong nên không chặn:

CompletableFuture.allOf(f1, f2, f3).join();
List<String> tat = Stream.of(f1, f2, f3).map(CompletableFuture::join).toList();

Khi có một cái hỏng:

  allOf ném lỗi: một cái hỏng
  nhưng g1 vẫn hoàn thành bình thường: ổn

allOf hỏng ngay nếu bất kỳ cái nào hỏng, nhưng các cái khác vẫn chạy và vẫn có kết quả. Muốn "lấy những cái thành công, bỏ qua cái hỏng" thì gắn exceptionally vào từng future trước khi đưa vào allOf.

Hay là đừng dùng nó nữa

Trên Java 21, phần lớn lý do dùng CompletableFuture đã biến mất. Bài 64 đã đo: sáu lời gọi HTTP song song bằng sendAsync mất 249 ms, bằng luồng ảo mất 47 ms — và mã luồng ảo là send() đồng bộ đọc từ trên xuống.

Khuyến nghị của tôi cho mã mới:

Tác vụ độc lập chạy song song rồi chờ hết → luồng ảo với Executors.newVirtualThreadPerTaskExecutor(). Đơn giản hơn nhiều, try/catch hoạt động bình thường, dấu vết ngăn xếp đọc được.

Chuỗi biến đổi thật sự, có nhánh và gộp → CompletableFuture vẫn đúng chỗ.

Đang bảo trì mã có sẵn → hiểu nó là đủ, đừng viết lại chỉ vì có công cụ mới.

Thử ba mươi giây

Tìm trong dự án của bạn supplyAsync( và runAsync(.

Với mỗi chỗ không truyền executor, hỏi: việc bên trong có chặn không — gọi mạng, truy vấn, đọc tệp? Có thì bạn đang chiếm dụng pool chung của cả JVM, và con số 31 giây ở đầu bài là thứ đang chờ bạn vào ngày lưu lượng tăng.

Ngày mai đi vào chính cái pool đã gây ra 31 giây ở trên: ForkJoinPool — cơ chế ăn trộm việc hoạt động ra sao, và viết một bài toán đệ quy song song rồi đo xem chia nhỏ tới đâu thì có lợi.