Concurrency patterns: Producer-Consumer, thread pool và CompletableFuture

Gọi ba kho hàng để so giá, tính phí vận chuyển, rồi gộp kết quả thành một hoá đơn: nếu viết tuần tự, mỗi lời gọi chờ lời gọi trước và thời gian phản hồi bằng tổng độ trễ. Java có sẵn bộ công cụ cho việc này, nhưng dùng sai thì lỗi rất khó thấy: pool kẹt, queue phình đến hết RAM, timeout không huỷ được công việc. Bài này đi từ nền tảng (Producer-Consumer với BlockingQueue, thread pool) đến CompletableFuture, mỗi phần có code JDK 21 chạy thật trên Apple M4 và output dán nguyên văn. Trọng tâm là những chỗ mà tài liệu Java và kinh nghiệm thực tế cho thấy dễ vấp nhất.

Bài nối tiếp Saga pattern và Transactional Outbox cho microservice, nơi các bước gọi service là tuần tự; ở đây ta xem cách chạy chúng đồng thời mà vẫn kiểm soát lỗi. Phần chịu lỗi cho từng lời gọi đã có ở Circuit Breaker, Retry và Bulkhead với Resilience4j.

Producer-Consumer: tách bên sinh việc khỏi bên xử lý

Mẫu Producer-Consumer đặt một hàng đợi giữa bên tạo công việc (producer) và bên xử lý (consumer) để hai bên chạy với tốc độ riêng. Trong JDK, BlockingQueue là cấu kiện chuẩn cho mẫu này. Theo tài liệu Java SE 21, put chờ khi hàng đợi đầy, take chờ khi hàng đợi rỗng, và hàng đợi có giới hạn làm chậm producer về tốc độ của consumer. Đó chính là backpressure. Interface này cũng không có thao tác đóng: cách thông dụng là producer đưa một đối tượng kết thúc ("poison pill") để consumer biết mà dừng.

BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);   // chỉ chứa 3 đơn

// producer
for (int i = 1; i <= orders; i++) queue.put("DH" + (100 + i));         // chặn nếu đầy
for (int i = 0; i < consumers; i++) queue.put(POISON);                 // mỗi consumer một viên

// consumer
String o;
while (!(o = queue.take()).equals(POISON)) {
    Thread.sleep(100);                                                  // mô phỏng xử lý
    processed.add(o + "@" + Thread.currentThread().getName());
}

Chạy 10 đơn, hai consumer, mỗi đơn xử lý 100 ms (Demo1ProducerConsumer):

Đã xử lý 10/10 đơn trong ~500 ms (làm tròn 100 ms)
Thời gian producer bị chặn vì queue đầy: > 100 ms (có backpressure)
consumer-1 xử lý 5 đơn, consumer-2 xử lý 5 đơn
Đơn nào cũng được xử lý đúng một lần: true

Mười đơn × 100 ms chia cho hai consumer ra khoảng 500 ms đúng như tính. Producer sinh việc nhanh hơn nhiều nhưng bị giữ lại vì queue chỉ có 3 chỗ; không có giới hạn đó, producer nhanh sẽ làm hàng đợi phình ra. Có ba điểm cần nhớ khi dùng mẫu này, trước khi sang CompletableFuture:

• Luôn đặt giới hạn: LinkedBlockingQueue không tham số có sức chứa gần như vô hạn, nên khi consumer chậm hơn producer thì bộ nhớ là thứ cạn trước.
• Mỗi consumer cần một poison pill: nếu chỉ gửi một viên cho hai consumer, consumer còn lại chờ take mãi và chương trình không thoát.
• Phân biệt put và offer(e, timeout, unit): put chờ vô hạn, còn offer có hạn chờ và trả về false để bạn quyết định bỏ, ghi log hay trả lỗi 503 cho client.

Thread pool: nền của mọi CompletableFuture bất đồng bộ

Tạo một Thread cho mỗi tác vụ vừa tốn bộ nhớ vừa không có điểm dừng khi quá tải. ThreadPoolExecutor gom các luồng và hàng đợi vào một đối tượng với bốn tham số: corePoolSize, maximumPoolSize, hàng đợi và chính sách từ chối. Quy tắc mà tài liệu ThreadPoolExecutor mô tả hay bị hiểu ngược: pool chỉ tạo thêm luồng vượt corePoolSize khi hàng đợi đã đầy, chứ không phải khi các luồng core bận. Hệ quả là với hàng đợi không giới hạn, maximumPoolSize hoàn toàn vô tác dụng.

Demo2ThreadPool dựng pool với core 2, max 4, queue 2, rồi gửi 8 tác vụ (mỗi tác vụ chạy 300 ms) theo hai chính sách từ chối:

--- AbortPolicy (mặc định) ---
T1 nhận: pool=1 thread, queue=0
T2 nhận: pool=2 thread, queue=0
T3 nhận: pool=2 thread, queue=1
T4 nhận: pool=2 thread, queue=2
T5 nhận: pool=3 thread, queue=2
T6 nhận: pool=4 thread, queue=2
T7 bị TỪ CHỐI (RejectedExecutionException)
T8 bị TỪ CHỐI (RejectedExecutionException)
--- CallerRunsPolicy ---
...
T7 nhận: pool=4 thread, queue=1
T8 nhận: pool=4 thread, queue=1
Thread chạy từng tác vụ: [T1@pool-2-thread-1, T2@pool-2-thread-2, T3@pool-2-thread-1, T4@pool-2-thread-3, T5@pool-2-thread-3, T6@pool-2-thread-4, T7@main, T8@pool-2-thread-4]

Tên luồng và kích thước queue ở T7, T8 của CallerRunsPolicy phụ thuộc thời điểm chạy (luồng pool có thể vừa lấy xong một tác vụ khỏi queue), nên mỗi lần chạy có thể lệch; cấu trúc và kết luận không đổi. Đọc output theo từng bước đúng như quy tắc: T1, T2 tạo luồng core; T3, T4 vào queue; chỉ khi queue đầy (T5, T6) pool mới nở lên 4 luồng; tới T7 thì cả luồng lẫn queue đều đầy. Với AbortPolicy, T7 và T8 bị ném RejectedExecutionException. Với CallerRunsPolicy, T7 chạy ngay trên luồng main (dòng cuối cho thấy T7@main): luồng gọi tự làm việc nên không thể gửi tiếp trong lúc đó, tạo ra backpressure tự nhiên.

Chú ý với CallerRunsPolicy trên server: luồng gọi có thể là luồng nhận request, nên khi quá tải chính luồng đó bị chiếm và server ngừng nhận kết nối mới. Đó là thứ ta muốn (nén tải về nguồn), nhưng phải biết để không ngạc nhiên. Cũng cần đặt tên luồng bằng ThreadFactory riêng để log và thread dump đọc được.

CompletableFuture: ghép các bước bất đồng bộ

Với Future cổ điển, lấy kết quả bằng get() là chặn luồng gọi, và không có cách ghép hai tác vụ mà không chặn. CompletableFuture (từ Java 8) cho phép khai báo chuỗi bước: khi kết quả sẵn sàng thì chạy bước tiếp theo. Các phương thức chính:

• supplyAsync(supplier, executor) và runAsync: khởi chạy tác vụ trên một executor.
• thenApply: biến đổi kết quả (như map); thenCompose: nối với một tác vụ bất đồng bộ khác phụ thuộc kết quả (như flatMap); thenCombine: ghép hai tác vụ độc lập.
• allOf chờ tất cả, anyOf lấy kết quả đầu tiên.
• exceptionally, handle, whenComplete cho xử lý lỗi; orTimeout và completeOnTimeout (từ Java 9) cho thời hạn.

Song song hoá các lời gọi độc lập bằng CompletableFuture

ExecutorService io = Executors.newFixedThreadPool(8);
List<CompletableFuture<Quote>> fs = List.of(
    CompletableFuture.supplyAsync(() -> call("kho-a", 520_000, 300), io),
    CompletableFuture.supplyAsync(() -> call("kho-b", 495_000, 300), io),
    CompletableFuture.supplyAsync(() -> call("kho-c", 510_000, 300), io));
CompletableFuture.allOf(fs.toArray(new CompletableFuture[0])).join();
List<Quote> par = fs.stream().map(CompletableFuture::join).toList();

Lưu ý allOf trả về CompletableFuture<Void>, không chứa kết quả: ta phải đọc từng future sau khi nó hoàn tất. Mỗi lời gọi giả lập 300 ms (Demo3Futures):

Tuần tự : ~900 ms, rẻ nhất = Quote[source=kho-b, price=495000]
Song song: ~300 ms, rẻ nhất = Quote[source=kho-b, price=495000]

Với CompletableFuture, ba lời gọi độc lập chạy đồng thời nên tổng thời gian gần bằng lời gọi chậm nhất, không phải tổng ba. Đây là con số đo trên máy chạy bài, độ trễ do sleep giả lập chứ không phải mạng thật.

Bước phụ thuộc và ghép kết quả

CompletableFuture<String> msg = CompletableFuture.supplyAsync(() -> call("kho-b", 495_000, 200), io)
    .thenCompose(q -> CompletableFuture.supplyAsync(() -> q.price() * 90 / 100, io))  // phụ thuộc: giảm 10%
    .thenCombine(shipping, (price, ship) -> "Tổng thanh toán = " + (price + ship) + " VND");

Tổng thanh toán = 475500 VND

Dùng thenCompose khi hàm của bạn tự trả về một CompletableFuture; nếu nhầm với thenApply bạn nhận CompletableFuture<CompletableFuture<T>>. shipping được khởi chạy trước đó nên chạy song song với chuỗi phía trên và thenCombine chỉ chờ cả hai xong.

Xử lý lỗi với CompletableFuture: ngoại lệ bị bọc

Khi một bước của CompletableFuture ném ngoại lệ, mọi bước phụ thuộc bị bỏ qua và ngoại lệ lan theo chuỗi. Ba phương thức khác nhau ở chỗ có đổi kết quả hay không: exceptionally thay lỗi bằng giá trị, handle nhận cả kết quả lẫn lỗi và trả giá trị mới, whenComplete chỉ chạy tác dụng phụ (như ghi log) và để kết quả nguyên vẹn. Chạy thật với một tác vụ ném IllegalStateException:

exceptionally: Quote[source=cache, price=500000]
whenComplete : chỉ ghi log, ex = CompletionException
handle       : lỗi: CompletionException <- kho-d mất kết nối
join() ném   : CompletionException, cause = IllegalStateException
get() ném    : ExecutionException, cause = IllegalStateException

Chi tiết đáng chú ý: ngoại lệ gốc IllegalStateException bị bọc trong CompletionException, kể cả khi bạn nhận nó trong handle hay whenComplete. Muốn lấy nguyên nhân thật phải gọi ex.getCause(), và vì lớp bọc có thể có hoặc không tuỳ vị trí trong chuỗi, code an toàn là mở lớp bọc trước khi so sánh kiểu. join() ném CompletionException (không kiểm tra), còn get() ném ExecutionException (có kiểm tra), cả hai cùng cause. Một chuỗi mà không ai gọi join, get hay xử lý lỗi sẽ nuốt ngoại lệ trong im lặng, nên mọi chuỗi cần một điểm kết thúc có xử lý lỗi.

Timeout của CompletableFuture: orTimeout không huỷ công việc đang chạy

Từ Java 9, orTimeout hoàn tất future với TimeoutException khi quá hạn, và completeOnTimeout hoàn tất với giá trị mặc định. Hai phương thức này tiện, nhưng có một hiểu lầm phổ biến:

Demo ghi thời điểm tác vụ gốc kết thúc ngay trong supplier bằng AtomicLong, rồi so với thời điểm timeout:

orTimeout        -> TimeoutException sau ~200 ms
tác vụ gốc đã xong chưa lúc timeout? false
tác vụ gốc thực sự kết thúc sau ~1000 ms (đo trong supplier)
completeOnTimeout -> Quote[source=mặc-định, price=999000]

Bên chờ nhận TimeoutException sau khoảng 200 ms, nhưng tác vụ gốc vẫn chạy và chỉ kết thúc ở mốc khoảng 1000 ms.

Tài liệu CompletableFuture của Java SE 21 nêu rằng cancel chỉ tương đương completeExceptionally(new CancellationException()) và tham số mayInterruptIfRunning bị bỏ qua, vì future không điều khiển được phép tính đang hoàn tất nó. Nghĩa là timeout chỉ giải phóng bên chờ, còn luồng đang thực thi lời gọi chậm vẫn bận tới khi xong. Muốn thật sự dừng, phải đặt timeout ở chính lời gọi (timeout của HTTP client, JDBC, socket). Timeout của CompletableFuture là lưới an toàn cho luồng chờ, không thay thế được timeout ở tầng I/O.

Bẫy lớn nhất của CompletableFuture: ForkJoinPool.commonPool

Nếu không truyền executor, các phương thức *Async của CompletableFuture chạy trên ForkJoinPool.commonPool(); tài liệu cũng ghi rằng nếu pool này không hỗ trợ song song từ hai trở lên thì mỗi tác vụ nhận một luồng mới. Hai hệ quả liên quan đến việc dùng chung: commonPool còn được parallel stream dùng, và theo tài liệu ForkJoinPool, pool không đảm bảo bù luồng khi gặp I/O chặn hoặc đồng bộ không được quản lý. Demo4CommonPool đo điều đó trên máy 10 lõi, với 40 lời gọi giả lập chặn 500 ms:

availableProcessors = 10
commonPool parallelism = 9
40 lời gọi chặn 500 ms trên commonPool : ~2500 ms
40 lời gọi chặn 500 ms trên pool riêng 40 thread: ~500 ms
tác vụ nhẹ (tổng 1..1000 = 500500) gửi vào commonPool đang bị chiếm: ~1900 ms

Parallelism của commonPool là 9 (số lõi trừ một, quan sát trên máy này), nên 40 lời gọi chặn chia thành khoảng 5 đợt, ra ~2500 ms. Pool riêng đủ 40 luồng chạy hết trong ~500 ms. Dòng cuối là hậu quả nguy hiểm nhất: một tác vụ tính tổng chỉ mất microsecond nhưng phải xếp hàng sau các lời gọi chặn và mất ~1900 ms. Mọi thứ khác trong JVM dùng commonPool (parallel stream, thư viện bên thứ ba) đều chịu cùng độ trễ. Quy tắc thực hành:

• Luôn truyền executor riêng cho tác vụ I/O, kích thước theo số lời gọi đồng thời mà nguồn bên ngoài chịu được, không theo số lõi.
• Tách pool theo loại việc (một pool cho DB, một cho HTTP ra ngoài) để nghẽn ở nơi này không lan sang nơi khác, cùng tinh thần bulkhead ở bài Resilience4j.
• Để commonPool cho việc tính toán thuần CPU, ngắn và không chặn.

thenApply hay thenApplyAsync: bước chạy ở luồng nào

pool-1-thread-1 -> thenApply chạy trên pool-1-thread-1 -> thenApplyAsync (mặc định) chạy trên ForkJoinPool.commonPool-worker-1

Bước thenApply chạy ngay trên luồng vừa hoàn tất bước trước (pool-1-thread-1), còn thenApplyAsync không truyền executor sẽ chuyển sang commonPool. Nếu bước trước đã hoàn tất sẵn khi gắn thenApply, bước này chạy ngay trên luồng gọi; thenApplyAsync luôn được giao cho executor. Hãy luôn truyền executor cho biến thể Async nếu bước có thể nặng hoặc chặn, vì nếu không nó rơi vào đúng cái bẫy ở trên.

Khi nào không nên dùng CompletableFuture

• Code chủ yếu tuần tự: chuỗi ba bước phụ thuộc nhau dùng thenCompose khó đọc và khó debug hơn ba dòng lệnh thường. Stack trace đứt đoạn giữa các luồng làm việc chẩn đoán khó hơn.
• Cần cancel thật sự: như đã thấy, cancel trên CompletableFuture không ngắt luồng; dùng ExecutorService.submit trả về Future với cancel(true), hoặc truyền cờ huỷ vào tác vụ.
• Luồng dữ liệu liên tục, có backpressure: dùng BlockingQueue như phần đầu, hoặc thư viện reactive; CompletableFuture chỉ biểu diễn một kết quả đơn.
• Hàng nghìn lời gọi chặn đồng thời: pool luồng nền tảng không scale; Java 21 có virtual thread cho đúng bài toán này, và series sẽ có bài riêng về nó.

Lỗi thường gặp

• Chặn trong commonPool: đã đo ở trên, nghẽn lan sang cả JVM.
• Hàng đợi không giới hạn trong Executors.newFixedThreadPool: hàng đợi nội bộ của nó không giới hạn, quá tải sẽ dẫn đến hết bộ nhớ chứ không bị từ chối. Với tải thật, tự dựng ThreadPoolExecutor với queue có giới hạn.
• Nuốt ngoại lệ: chuỗi CompletableFuture không có join, get, exceptionally hay handle cuối cùng sẽ mất lỗi.
• Gọi join() trong một bước đang chạy trên cùng pool: khi pool nhỏ, các luồng đều chờ nhau và pool kẹt (thread starvation); chờ ở biên của chuỗi chứ không ở giữa.
• Quên shutdown executor tự tạo: luồng không phải daemon giữ cho JVM không thoát.
• Tin rằng orTimeout dừng được tác vụ: nó chỉ giải phóng bên chờ.
• Dùng allOf mà bỏ sót lỗi: nếu một CompletableFuture lỗi, allOf cũng hoàn tất bất thường; chuẩn bị xử lý và quyết định có chấp nhận kết quả một phần hay không.

Checklist

• Producer-Consumer: queue có giới hạn, mỗi consumer một poison pill, dùng offer có timeout nếu không muốn chặn vô hạn.
• Thread pool: tự khai báo ThreadPoolExecutor, hiểu quy tắc core, queue, max; chọn chính sách từ chối có chủ đích; đặt tên luồng.
• CompletableFuture: truyền executor riêng cho I/O; thenCompose cho bước phụ thuộc, thenCombine cho bước độc lập.
• Luôn có điểm kết thúc xử lý lỗi và nhớ ngoại lệ bị bọc trong CompletionException.
• Timeout đặt ở tầng I/O; orTimeout chỉ là lưới an toàn cho bên chờ.
• Đo thay vì đoán: thời gian chạy, kích thước hàng đợi và số luồng cần theo dõi bằng metric.

Tóm lại, các mẫu đồng thời của Java đều xoay quanh một câu hỏi: công việc đang chờ ở đâu và ai giới hạn nó. BlockingQueue giới hạn bằng sức chứa, thread pool giới hạn bằng số luồng và chính sách từ chối, còn CompletableFuture lo việc ghép kết quả nhưng không tự giải quyết vấn đề giới hạn. Toàn bộ mã nằm trong code/s6-29, chạy bằng JDK 21 (LTS) trên Apple M4 (16 GB RAM); các con số thời gian là giá trị làm tròn của một lần chạy.

Bài tiếp theo

Sau chuyện đồng thời, ta quay lại mẫu thiết kế kinh điển và xem Java 21 thay đổi cách viết nó thế nào. Mời bạn đọc tiếp Visitor pattern thời Java 21: sealed interface và pattern matching.