Blocking queue & producer–consumer: tách nhịp sản xuất–tiêu thụ

TL;DR: BlockingQueue là hàng đợi thread-safe biết chờ: take() chặn khi rỗng, put() chặn khi đầy — hai năng lực làm nên xương sống của mẫu producer–consumer. Hàng đợi bounded biến put/offer thành van backpressure: producer nhanh hơn consumer kéo dài sẽ bị ghì lại, thay vì để hàng đợi phình tới OutOfMemoryError. API có bốn kiểu hành xử cho mỗi thao tác — ném exception, trả tín hiệu, chặn vô hạn, chặn có timeout — chọn theo điều bạn muốn khi không làm được ngay. Object đi qua hàng đợi còn được safe publication miễn phí, và shutdown sạch bằng poison pill (N consumer cần N viên) hoặc interrupt. Pitfall lớn nhất: hàng đợi unbounded không xóa giới hạn hệ thống, nó chỉ giấu giới hạn đó đi rồi đổi vài request bị từ chối lấy một cú OOM lúc tải đỉnh.

1. Vì sao producer và consumer cần một hàng đợi biết chờ?

Các concurrent collection ở bài trước — ConcurrentHashMap, CopyOnWriteArrayList, ConcurrentLinkedQueue — thả vào là dùng, tự lo phần đồng bộ khó. Nhưng chúng có một điểm chung: lấy mà không có gì để lấy thì trả về ngay, poll() trên hàng đợi rỗng cho null lập tức. Trong rất nhiều bài toán, chính cái chờ đó mới là thứ ta cần.

Mẫu producer–consumer tách một hoạt động thành hai vai — bên tạo ra công việc và bên thực hiện công việc — nối nhau bằng một hàng đợi ở giữa. Producer bỏ việc vào rồi quay đi làm tiếp; consumer lấy việc ra rồi xử lý. Hai bên không gọi trực tiếp nhau, không cần biết nhau, thậm chí không cần cùng số lượng. Hình ảnh gần nhất là quầy pha chế quán cà phê đông khách: thu ngân kẹp phiếu order vào kẹp, barista lấy phiếu trên cùng rồi pha — cao điểm thì kẹp dày lên, vắng khách thì mỏng đi, hệ thống tự cân bằng quanh cái kẹp mà không cần điều phối thủ công.

→

→

Tách rời như vậy không chỉ cho gọn: hai bên được suy luận, kiểm thử và điều chỉnh độc lập — tăng số consumer mà không động vào producer. Và quan trọng nhất với hệ thống chịu tải, hàng đợi ở giữa trở thành nơi duy nhất để áp đặt giới hạn, thay vì rải kiểm soát tải khắp nơi.

Điều thú vị xảy ra khi hai bên không cùng tốc độ — thực tế gần như luôn vậy. Producer nhanh hơn kéo dài thì hàng đợi dài ra, và câu hỏi sống còn là: dài đến đâu thì dừng? Consumer nhanh hơn thì hàng đợi thường rỗng, và nó phải chờ việc mới sao cho không đốt CPU. Hai tình huống đối xứng đó — chờ khi đầy, chờ khi rỗng — chính là hai năng lực BlockingQueue đóng gói sẵn. Tự viết chúng bằng List cộng wait/notify là cái bẫy kinh điển: quên kiểm tra điều kiện trong vòng while, dùng notify thay vì notifyAll, để lọt missed signal.

2. BlockingQueue API: bốn kiểu hành xử

BlockingQueue<E> nằm trong java.util.concurrent, có từ Java 5, mở rộng Queue với các thao tác biết chờ. Điều đáng nhớ: mỗi thao tác cơ bản — thêm vào, lấy ra — đều có nhiều biến thể tùy theo bạn muốn nó hành xử thế nào khi không làm được ngay.

BlockingQueue<Task> queue = new ArrayBlockingQueue<>(1000);

// Producer: chan neu queue day → backpressure tu nhien
queue.put(task);

// Consumer: chan neu queue rong → khong busy-wait, ngu toi khi co viec
Task task = queue.take();

put và take là cặp đơn giản và mạnh nhất: toàn bộ phần phối hợp "đầy thì chặn, rỗng thì chặn, có biến động thì đánh thức đúng bên" nằm gọn trong hai lời gọi đó, và thread nằm chờ không tốn CPU. Cả hai đều ném InterruptedException, vì khi đang chặn chúng phản hồi interrupt — điểm ta sẽ quay lại ở phần shutdown.

Cơ chế bên dưới cái chờ này không huyền bí: nó chính là pattern hai Condition trên một ReentrantLock mà bài ReentrantLock & Condition đã dựng tay. Mở source ArrayBlockingQueue, bạn thấy đúng bộ ba quen thuộc: một ReentrantLock canh mảng, condition notEmpty cho consumer nằm chờ, condition notFull cho producer nằm chờ. put vào hàng đầy thì notFull.await(); take rút bớt một phần tử thì notFull.signal() đánh thức đúng phía producer, đối xứng cho chiều ngược lại. BlockingQueue không thay thế kiến thức về lock và condition — nó là kiến thức đó được đóng gói sẵn.

Khi chặn vô hạn quá rủi ro — servlet thread không nên nằm chờ mãi vì còn phải trả response cho client — thì dùng biến thể có timeout. offer(task, 100, MILLISECONDS) cố bỏ vào trong tối đa 100ms; vẫn không có chỗ thì trả false và ta chủ động fail-fast thay vì treo.

Một chi tiết hay bị bỏ sót: đặt một object vào BlockingQueue còn lo luôn safe publication cho object đó, vì thao tác bỏ vào một thread-safe collection thiết lập quan hệ happens-before với thao tác lấy ra — chính cái nền mà bài Immutability đã dựng. Object dựng đúng cách, sau khi đi qua hàng đợi, hiển thị lành lặn với consumer mà không cần đồng bộ thêm: hàng đợi vừa truyền dữ liệu, vừa truyền memory visibility.

3. Bounded, unbounded và các biến thể

3.1 ArrayBlockingQueue và LinkedBlockingQueue

ArrayBlockingQueue được nâng đỡ bởi một mảng kích thước cố định chốt lúc khởi tạo — nó luôn bounded. LinkedBlockingQueue được nâng đỡ bởi node liên kết: bounded nếu truyền capacity, unbounded nếu không — khi đó trần mặc định là Integer.MAX_VALUE, lớn tới mức thực tế coi như vô hạn.

BlockingQueue<Task> array     = new ArrayBlockingQueue<>(1000);    // tran cung 1000
BlockingQueue<Task> bounded   = new LinkedBlockingQueue<>(1000);   // co tran
BlockingQueue<Task> unbounded = new LinkedBlockingQueue<>();       // "vo han" — coi chung

ArrayBlockingQueue cấp phát mảng một lần và dùng một khóa cho cả hai đầu, nên đơn giản và có dấu chân bộ nhớ ổn định. LinkedBlockingQueue cấp phát node theo nhu cầu và tách khóa đầu lấy ra khỏi khóa đầu bỏ vào, nên dưới tải cao throughput thường nhỉnh hơn vì hai đầu ít tranh khóa. Nhưng đó là tối ưu hiệu năng; quyết định quan trọng hơn nằm ở chữ "bounded".

3.2 Bounded nghĩa là có backpressure

Hàng đợi bounded biến put thành một van an toàn. Khi đầy, producer gọi put bị chặn lại — nó không thể chạy nhanh hơn consumer kéo dài, vì đến một điểm chính hàng đợi sẽ ghì nó lại. Đây là backpressure ở dạng tinh khiết nhất: không phải viết một dòng logic kiểm soát tải nào, chỉ cần chọn capacity hợp lý. Quầy cà phê cũng vậy — kẹp order đầy thì thu ngân buộc phải ngừng nhận khách, khó chịu nhưng đó là điều cứu cả hệ thống.

Producer không cần dòng code nào hỏi "consumer theo kịp chưa": chính lời gọi put bị chặn là câu trả lời, và khoảnh khắc consumer rút bớt một phần tử, producer được đánh thức đi tiếp.

3.3 Cái bẫy unbounded: backpressure biến mất, thay bằng OOM

Một LinkedBlockingQueue không capacity nghe có vẻ tiện — không bao giờ phải lo put bị chặn. Chính sự tiện đó là cái bẫy: backpressure biến mất hoàn toàn, và producer nhanh hơn consumer kéo dài thì hàng đợi cứ phình ra mãi.

Hàng đợi đó sống trên heap. Mỗi phần tử chưa xử lý là một object còn reachable, garbage collector không thu được: hàng đợi phình tới đâu, heap căng tới đó, cho tới khoảnh khắc OutOfMemoryError và toàn bộ ứng dụng sụp. Lỗi này độc ác vì nó im lặng suốt giai đoạn lành mạnh — tải thấp thì hai bên cân bằng, hàng đợi luôn ngắn, mọi test pass — rồi chỉ lộ đúng lúc tải đạt đỉnh, khi ta cần hệ thống đứng vững nhất. Unbounded không loại bỏ giới hạn của hệ thống; nó giấu giới hạn đó đi, rồi đổi một sự cố nhẹ (vài request bị từ chối) lấy một sự cố nặng: cả tiến trình chết. Vì vậy, gần như mọi hàng đợi trên đường đi của tải sản xuất nên là bounded — hãy hỏi "khi đầy thì sao?" lúc thiết kế, đừng để một cú crash lúc 2 giờ sáng trả lời hộ.

3.4 Các biến thể cho ràng buộc đặc thù

java.util.concurrent còn vài cài đặt giải đúng những bài toán riêng — biết chúng tồn tại để khỏi tự dựng lại:

• PriorityBlockingQueue — lấy ra theo độ ưu tiên (Comparator hoặc Comparable) thay vì đến trước phục vụ trước. Hợp khi việc không bình đẳng: request VIP đến sau vẫn xử lý trước. Nó unbounded về mặt logic, vẫn phải canh chừng OOM.
• SynchronousQueue — không có dung lượng nào cả, kể cả một chỗ: mỗi put phải chờ một take tương ứng và ngược lại. Nó không lưu trữ phần tử mà chỉ làm điểm hẹn để trao tay trực tiếp (direct handoff). Đây chính là cơ chế bên trong Executors.newCachedThreadPool(): task chỉ vào được pool khi có ngay thread rảnh nhận, nếu không pool tạo thread mới.
• DelayQueue — mỗi phần tử cài đặt Delayed để nói khi nào nó "chín"; take chỉ trả phần tử khi delay đã hết, trước đó dù hàng đợi không rỗng vẫn chờ. Hợp cho cache có TTL, retry có backoff, task hẹn giờ.
• LinkedTransferQueue (Java 7) — dạng lai, bổ sung transfer(e): producer có thể chọn chờ tới khi consumer thực sự nhận phần tử của mình (giống SynchronousQueue), nhưng vẫn put không chờ được như hàng đợi thường. Lock-free ở fast path.
• BlockingDeque (LinkedBlockingDeque, từ Java 6) — chặn hai đầu: bỏ vào và lấy ra được ở cả đầu lẫn cuối. Nền cho mẫu work stealing, nơi worker rảnh "trộm" việc từ đầu kia deque của worker khác.

Phần lớn thời gian, một ArrayBlockingQueue hay LinkedBlockingQueue bounded là đủ. Các biến thể trên là để khi bài toán có ràng buộc đặc thù — ưu tiên, handoff, thời gian, hai đầu — ta nhận ra ngay đã có công cụ sẵn.

4. Shutdown sạch: poison pill và interrupt

Một pipeline producer–consumer chạy được thì dễ; cho nó dừng sạch mới là phần lộ ra tay nghề. Consumer điển hình là vòng lặp while quanh take, mà take chặn vô hạn khi hàng đợi rỗng. Vậy làm sao bảo một consumer đang nằm chờ ở take rằng "hết việc rồi, về đi"?

4.1 Poison pill

Cách kinh điển nhất là poison pill: một phần tử đặc biệt, không phải việc thật, mà là tín hiệu "dừng lại". Producer bỏ nó vào hàng đợi như phần tử cuối cùng; consumer take ra mà thấy nó thì thoát vòng lặp.

final Task POISON = new Task();          // sentinel, khong mang du lieu that

// Consumer
void consume() throws InterruptedException {
    while (true) {
        Task t = queue.take();          // chan toi khi co phan tu
        if (t == POISON) break;         // gap pill → dung sach
        process(t);
    }
}
queue.put(POISON);                      // producer: rai pill khi het viec that

Cái hay của poison pill là nó tôn trọng thứ tự hàng đợi: consumer xử lý hết việc thật nằm trước pill rồi mới dừng. Nếu có N consumer, ta cần N viên, vì mỗi viên chỉ dừng đúng một consumer lấy được nó; và producer phải ngừng bỏ việc thật trước khi rải pill, nếu không pill có thể bị lấy trong khi việc thật vẫn còn phía sau. Poison pill chỉ gọn khi số producer và consumer đều biết trước; khi số lượng động, interrupt tự nhiên hơn.

4.2 Interrupt khi take đang chặn

Vì take và put ném InterruptedException khi thread bị interrupt lúc đang chặn, ta có thể dừng consumer bằng cách interrupt thread của nó: lời gọi take đang nằm chờ bật dậy bằng exception đó, và consumer dùng nó làm tín hiệu thoát.

Điều quan trọng là xử lý InterruptedException cho đúng, theo nguyên tắc cooperative cancellation đã học ở bài Thread API và vòng đời: hoặc ném tiếp lên trên, hoặc khôi phục cờ interrupt bằng Thread.currentThread().interrupt() trước khi thoát, để tầng trên còn biết thread đã bị yêu cầu dừng. Nuốt im lặng nó là một trong những lỗi concurrency tệ nhất vì làm mất tín hiệu hủy.

Capstone TicketFlow dùng đúng pattern này. Hàng đợi ở đây mang FutureTask — "gói việc" vừa chạy được vừa giữ kết quả trả về cho người gửi; định nghĩa đầy đủ ở mục 5.

// BookingWorker — consumer side, capstone v2
@Override
public void run() {
    while (running) {
        FutureTask<BookingResult> task;
        try {
            task = queue.take();                       // chan toi khi co task
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();        // khoi phuc co roi thoat sach
            return;
        }
        task.run();
    }
}

Worker chạy tới khi cờ running (một volatile boolean) bị tắt, hoặc tới khi thread bị interrupt giữa lúc đang take — cả hai con đường đều dẫn ra khỏi vòng lặp có trật tự. Cờ running là volatile để thay đổi của thread gọi stop() hiển thị kịp với worker — đúng cái visibility mà bài volatile & synchronized đã nhấn mạnh.

5. Capstone: pipeline đặt vé qua BlockingQueue

Ở các version trước, một servlet thread gọi thẳng BookingService.book và nằm chờ business logic chạy xong — trộn lẫn việc nhận request với việc xử lý request, không có chỗ áp giới hạn tải. Version v2 tách hai vai đó bằng một BlockingQueue bounded ở giữa.

Phía producer là BookingGateway: servlet thread gọi submit, gateway gói business logic vào một FutureTask, đẩy vào hàng đợi, rồi trả Future về cho caller — caller chờ kết quả qua future.get(timeout).

// BookingGateway — producer side, capstone v2
public Future<BookingResult> submit(BookingRequest req) {
    FutureTask<BookingResult> task = new FutureTask<>(() -> {
        try { return BookingResult.ok(service.book(req.eventId(), req.userId())); }
        catch (SoldOutException e) { return BookingResult.fail("Sold out: " + req.eventId()); }
    });
    boolean accepted;
    try {
        accepted = queue.offer(task, offerTimeoutMillis, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RejectedException("Interrupted while submitting");
    }
    if (!accepted) throw new RejectedException("Queue full, try again");   // backpressure thanh tin hieu
    return task;
}

Điểm thiết kế đáng chú ý nhất: gateway dùng offer có timeout chứ không phải put, vì servlet thread không nên nằm chờ vô hạn khi còn nợ client một response. Gateway chỉ cho hàng đợi tối đa offerTimeoutMillis để nhận task; hết thời gian mà vẫn đầy, offer trả false, và gateway biến sự cố "đầy" đó thành một RejectedException rõ ràng — câu trả lời "thử lại sau" gửi ngược về client. Backpressure ở đây là tín hiệu fail-fast phía thượng nguồn nhận được ngay.

Đó chính là khác biệt bounded/unbounded thể hiện thành code thật: vì hàng đợi có trần, đầy là một trạng thái có thật mà gateway phải xử lý — bằng cách từ chối lịch sự thay vì để task chất đống. Nếu unbounded, offer luôn thành công, RejectedException không bao giờ ném, và hệ thống lặng lẽ trôi về phía OOM dưới tải đỉnh.

Phía consumer là BookingWorker ở mục 4.2: một hoặc nhiều worker thread cùng take rồi chạy FutureTask. Vì nhiều worker cùng rút từ một BlockingQueue thread-safe, ta tăng số worker để nâng throughput mà không động gì tới gateway. BookingService bên trong vẫn giữ invariant "không bán vượt capacity": hàng đợi không thay đổi tính đúng đắn của business logic, nó chỉ định hình lại nhịp mà logic đó được gọi. Và vì FutureTask tự hấp thụ mọi exception rồi lan ra qua future.get(), một task hỏng không giết chết worker.

Toàn cảnh v2 vì thế là producer–consumer sách giáo khoa: phần đồng bộ hóa khó nhất — đầy thì chặn, rỗng thì chờ, trao tay an toàn về memory — nằm trọn trong hàng đợi mà ta không viết một dòng wait/notify nào.

6. Liên hệ các bài khác

• Thread API và vòng đời — interrupt là cơ chế shutdown thứ hai của pipeline; bài đó giải thích vì sao phải khôi phục cờ interrupt.
• Immutability — safe publication mà hàng đợi cho "miễn phí" được dựng nền ở đó; phần tử immutable đi qua queue là tổ hợp an toàn nhất.
• ReentrantLock & Condition — ArrayBlockingQueue chính là pattern một lock + hai condition (notFull/notEmpty) của bài đó, đóng gói sẵn.
• Synchronizers — họ công cụ điều phối tiến độ không kèm dữ liệu; BlockingQueue cũng là một synchronizer kiêm vai chứa dữ liệu.
• Executor & thread pool — thread pool là producer–consumer hoàn chỉnh: submit (producer), worker thread (consumer), work queue ở giữa — đúng kiến trúc capstone vừa dựng tay.

7. 📚 Deep Dive Oracle

8. Tóm tắt

• BlockingQueue tách nhịp sản xuất khỏi nhịp tiêu thụ: lấy khi rỗng thì chặn, bỏ vào khi đầy thì chặn.
• API có bốn kiểu hành xử cho mỗi thao tác: ném exception (add/remove), trả tín hiệu (offer/poll), chặn vô hạn (put/take), chặn có timeout.
• Bounded là nơi backpressure sống. Unbounded chỉ giấu giới hạn của hệ thống đi, rồi đổi một sự cố nhẹ lấy một cú OOM lúc tải đỉnh. Mặc định nên bounded.
• Biến thể giải bài toán riêng: PriorityBlockingQueue cho việc không bình đẳng, SynchronousQueue cho handoff trực tiếp, DelayQueue cho lập lịch theo thời gian, LinkedTransferQueue khi cần biết chắc có người nhận, BlockingDeque cho work stealing.
• Shutdown sạch dùng poison pill (gọn khi số bên cố định) hoặc interrupt (tự nhiên khi số bên động) — đừng nuốt im lặng InterruptedException.
• Bỏ object vào hàng đợi cũng lo luôn safe publication: object hiển thị lành lặn với consumer, không cần đồng bộ thêm.

BlockingQueue điều phối dữ liệu. Nhưng còn một họ công cụ khác chỉ lo điều phối tiến độ: cho các thread chờ nhau, giới hạn số thread cùng vào một vùng, hay hẹn gặp tất cả tại một điểm chung. Đó là các synchronizer — chủ đề bài kế tiếp.

9. Tự kiểm tra

Bài tiếp theo: Synchronizers — điều phối tiến độ giữa nhiều thread