您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Java 多線程核心組件 BlockingQueue 深度解析：原理、五大實現、生產消費者最佳實踐

編輯：Java 綜合教學

Java 多線程核心組件 BlockingQueue 深度解析：原理、五大實現、生產消費者最佳實踐

BlockingQueue 是 Java 併發包中線程安全、支持阻塞操作的隊列介面，完美契合生產者-消費者模式，避免手動 wait/notify 易錯誤。

五大核心實現對比

實現類	底層結構	容量	特點	適用場景
ArrayBlockingQueue	數組	有界（構造時指定）	公平/非公平鎖可選，性能穩定	固定緩衝區、高併發穩定吞吐
LinkedBlockingQueue	鏈表	可選有界（預設 Integer.MAX_VALUE）	吞吐量高，動態擴容	通用生產消費、日誌異步寫入
PriorityBlockingQueue	堆（優先隊列）	無界	按優先級出隊，非 FIFO	任務調度、優先級作業處理
DelayQueue	優先隊列	無界	元素延遲激活，到期才可取	定時任務、緩存過期清理、重試機制
SynchronousQueue	無緩衝	0（直接移交）	放/取必須配對，極高併發傳遞	線程池直接提交、雙邊交換

核心方法語義分類

最佳實踐：優先使用 offer/poll/peek 非阻塞版本，通過返回值判斷成敗；阻塞版本用於明確需要等待的業務邏輯。

原理剖析：以 ArrayBlockingQueue 為例

public class ArrayBlockingQueue extends AbstractQueue implements BlockingQueue, Serializable {
    final Object[] items;        // 環形數組
    int putIndex;                // 寫索引
    int takeIndex;               // 讀索引
    int count;                   // 元素數量
    final ReentrantLock lock;    // 獨佔鎖
    private final Condition notEmpty; // 非空條件
    private final Condition notFull;  // 非滿條件

    public void put(E e) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (count == items.length) notFull.await(); // 滿則等待
            items[putIndex] = e;
            putIndex = (putIndex + 1) % items.length;
            count++;
            notEmpty.signal(); // 喚醒消費者
        } finally { lock.unlock(); }
    }

    public E take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (count == 0) notEmpty.await(); // 空則等待
            E e = items[takeIndex];
            items[takeIndex] = null;
            takeIndex = (takeIndex + 1) % items.length;
            count--;
            notFull.signal(); // 喚醒生產者
            return e;
        } finally { lock.unlock(); }
    }
}

經典應用場景代碼

1. 兩線程互斥等待（交替打印）

BlockingQueue q = new ArrayBlockingQueue<>(3);
// 生產者
new Thread(() -> { while(true){ q.put(1); System.out.println("放入"); }}).start();
// 消費者
new Thread(() -> { while(true){ q.take(); System.out.println("取出"); }}).start();

2. 主從線程交替執行（替代 wait/notify）

class Business {
    BlockingQueue q1 = new ArrayBlockingQueue<>(1);
    BlockingQueue q2 = new ArrayBlockingQueue<>(1);
    { q2.put(1); } // 初始阻塞 q2

    void main(int i) throws InterruptedException {
        q1.put(1); // 阻塞自己
        for(int j=0;j<100;j++) System.out.println("main "+j);
        q2.take(); // 喚醒子線程
    }
    void sub(int i) throws InterruptedException {
        q2.put(1);
        for(int j=0;j<10;j++) System.out.println("sub "+j);
        q1.take();
    }
}

注意：BlockingQueue 內部已同步，切勿在方法上再加 synchronized，否則死鎖。

3. 標準生產者-消費者模板（API 文檔範例）

class Producer implements Runnable {
    private final BlockingQueue q;
    Producer(BlockingQueue q){ this.q=q; }
    public void run(){ try{ while(true) q.put(produce()); }catch(InterruptedException e){} }
    Object produce(){ return new Object(); }
}
class Consumer implements Runnable {
    private final BlockingQueue q;
    Consumer(BlockingQueue q){ this.q=q; }
    public void run(){ try{ while(true) consume(q.take()); }catch(InterruptedException e){} }
    void consume(Object x){}
}
class Setup {
    void main(){
        BlockingQueue q = new LinkedBlockingQueue();
        new Thread(new Producer(q)).start();
        new Thread(new Consumer(q)).start();
        new Thread(new Consumer(q)).start();
    }
}