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, ConcurrentLinkedQueue — thả vào là dùng, tự lo phần đồng bộ khó. Nhưng chúng có điểm chung: lấy mà không có gì 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 việc và bên xử lý việc — nối nhau bằng một hàng đợi ở giữa. Hai bên không gọi trực tiếp nhau, không cần biết nhau, 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.

→

→

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, và hàng đợi ở giữa trở thành nơi duy nhất để áp đặt giới hạn tải. Nhưng hai bên gần như không bao giờ cùng tốc độ — hai tình huống đối xứng nảy sinh: chờ khi đầy, chờ khi rỗng. Đó chính là hai năng lực BlockingQueue đóng gói sẵn; tự viết bằng List cộng wait/notify là cái bẫy kinh điển đã mổ ở bài 09.

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 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, thread nằm chờ không tốn CPU. Cả hai ném InterruptedException — điểm ta quay lại ở phần shutdown.

Cơ chế bên dưới không huyền bí: source ArrayBlockingQueue chính là pattern một ReentrantLock + hai Condition (notEmpty/notFull) mà bài ReentrantLock & Condition đã dựng tay. BlockingQueue không thay thế kiến thức lock/condition, nó là kiến thức đó đóng gói sẵn. Còn khi chặn vô hạn quá rủi ro, biến thể timeout ở hàng cuối bảng là van xả — capstone ở mục 5 dùng đúng nó.

Một chi tiết hay bị bỏ sót: đặt object vào BlockingQueue còn lo luôn safe publication — bỏ vào và lấy ra nối nhau bằng happens-before (Immutability), nên object dựng đúng cách 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 dùng một khóa cho cả hai đầu; LinkedBlockingQueue tách khóa hai đầu nên throughput dưới tải cao thường nhỉnh hơn — 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ý.

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.

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ờ lo put bị chặn. Chính sự tiện đó là cái bẫy: backpressure biến mất, producer nhanh hơn consumer kéo dài thì hàng đợi phình mãi — mỗi phần tử chưa xử lý là một object còn reachable, heap căng dần tới OutOfMemoryError. Lỗi này im lặng lúc tải thấp (hàng đợi luôn ngắn, mọi test pass) rồi lộ đúng lúc tải đỉnh. Unbounded không loại bỏ giới hạn hệ thống; nó giấu đ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). Hãy hỏi "khi đầy thì sao?" lúc thiết kế.

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:

Phần lớn thời gian, một queue bounded thường là đủ; bảng trên để khi bài toán có ràng buộc đặc thù, ta nhận ra ngay đã có công cụ sẵn.

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

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

Poison pill 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 consumer cần N viên (mỗi viên dừng đúng một consumer), và producer phải ngừng bỏ việc thật trước khi rải pill — pill phải là phần tử cuối cùng. Nó chỉ gọn khi số bên biết trước; số lượng động thì interrupt tự nhiên hơn.

4.2 Interrupt khi take đang chặn

Vì take và put ném InterruptedException khi bị interrupt lúc đang chặn, ta dừng consumer bằng cách interrupt thread của nó: lời gọi take đang chờ bật dậy bằng exception đó, consumer dùng làm tín hiệu thoát. Điều quan trọng là xử lý đúng theo cooperative cancellation ở 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 — nuốt im lặng 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ề; đị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, servlet thread gọi thẳng BookingService.book — trộ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: BookingGateway (producer) gói business logic vào FutureTask, đẩy vào hàng đợi, trả Future cho caller chờ 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 được treo vô hạn khi còn nợ client một response. Hết offerTimeoutMillis mà hàng đợi vẫn đầy, offer trả false, gateway biến trạng thái "đầy" thành RejectedException rõ ràng — "thử lại sau" gửi ngược về client thay vì để task chất đống về phía OOM.

Phía consumer là BookingWorker (mục 4.2): nhiều worker cùng take rồi chạy FutureTask — tăng worker là nâng throughput mà không động tới gateway. FutureTask hấp thụ exception rồi lan qua future.get() nên một task hỏng không giết worker; hàng đợi chỉ định hình lại nhịp gọi logic, không đổi tính đúng đắn, 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ử 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 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ể chuyên biệt (Priority/Synchronous/Delay/Transfer/Deque) dùng khi có ràng buộc riêng.
• 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.

Bài kế tiếp chuyển sang họ công cụ chỉ điều phối tiến độ (không kèm dữ liệu) — cho thread chờ nhau, giới hạn số thread vào một vùng, hẹn gặp tại một điểm chung: các synchronizer.

9. Tự kiểm tra

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