您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 併發容器：BlockingQueue 阻塞隊列家族全解析與生產者消費者實戰

編輯：Java 相關知識

BlockingQueue 核心價值

解決多線程環境下**高效、安全**的資料傳遞。內建阻塞機制：

隊列空 → 消費者 take() 阻塞等待
隊列滿 → 生產者 put() 阻塞等待

開發者無需手動處理 wait/notify、鎖、條件變量。

核心 API 速查

七大實作類詳解

方法	行為	阻塞?	超時支援?
offer(e)	嘗試加入，失敗返回 false	否	否
offer(e, timeout, unit)	超時前嘗試加入	是	是
put(e)	加入，滿則阻塞等待	是	否
poll()	取走隊首，空返回 null	否	否
poll(timeout, unit)	超時前嘗試取走	是	是
take()	取走隊首，空則阻塞等待	是	否
drainTo(c)	批量取出所有可用元素	否	否

類別	底層結構	容量	鎖機制	特點/適用場景
ArrayBlockingQueue	陣列	固定(建構式指定)	單一鎖	簡單、公平鎖可選、無額外物件創建、GC 友好
LinkedBlockingQueue	鏈表	可選(預設 Integer.MAX_VALUE)	雙鎖(分離)	高並發、生產消費可真正並行、預設無界危險
PriorityBlockingQueue	堆(優先隊列)	無界	單一公平鎖	元素需 Comparable/Comparator、僅阻塞消費者
DelayQueue	堆(延遲隊列)	無界	單一鎖	元素需 Delayed、僅阻塞消費者、定時任務/連接超時管理
SynchronousQueue	無緩衝	0	公平/非公平	直接交付、生產消費配對、極高吞吐低延遲
LinkedTransferQueue	鏈表	無界	CAS+自旋	JDK7+、transfer()語義、比 SynchronousQueue 功能豐富
LinkedBlockingDeque	雙向鏈表	可選	雙鎖	雙端隊列、支援 FIFO/LIFO、工作竊取演算法

⚠️ 關鍵陷阱

LinkedBlockingQueue 無界預設：生產快於消費 → OOM。必須指定容量。
PriorityBlockingQueue/DelayQueue 無界：同理需監控生產速率。
SynchronousQueue 公平 vs 非公平：公平用 FIFO 隊列+公平鎖；非公平(預設)用 LIFO 隊列+非公平鎖，可能導致飢餓。

完整生產者消費者實戰

public class BlockingQueueTest {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(10); // 必須有界！
        ExecutorService es = Executors.newCachedThreadPool();

        // 3 生產者
        for (int i = 0; i < 3; i++) es.execute(new Producer(queue));
        // 1 消費者
        es.execute(new Consumer(queue));

        Thread.sleep(10_000);
        // 優雅關閉
        Producer.stopAll();
        Thread.sleep(2000);
        es.shutdown();
    }
}

class Producer implements Runnable {
    private static final AtomicInteger count = new AtomicInteger();
    private static volatile boolean running = true;
    private final BlockingQueue<String> queue;
    private final Random r = new Random();

    public Producer(BlockingQueue<String> q) { this.queue = q; }
    public static void stopAll() { running = false; }

    @Override
    public void run() {
        while (running) {
            try {
                String data = "data:" + count.incrementAndGet();
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                    System.out.println("入隊超時：" + data);
                }
                Thread.sleep(r.nextInt(1000));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

class Consumer implements Runnable {
    private final BlockingQueue<String> queue;
    private final Random r = new Random();

    public Consumer(BlockingQueue<String> q) { this.queue = q; }

    @Override
    public void run() {
        try {
            while (true) {
                String data = queue.poll(2, TimeUnit.SECONDS);
                if (data == null) break; // 2秒無數據視為結束
                System.out.println("消費：" + data);
                Thread.sleep(r.nextInt(1000));
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

選型指南

**一般生產消費、需有界** → ArrayBlockingQueue (簡單) 或 LinkedBlockingQueue(capacity) (高並發)
**優先級處理** → PriorityBlockingQueue (注意無界風險)
**定時/延遲任務** → DelayQueue
**直接交付、極低延遲** → SynchronousQueue (公平模式)
**工作竊取、雙端操作** → LinkedBlockingDeque