您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 程式設計入門

Java 並發核心：阻塞隊列完整指南 - 原理、實現、生產者消費者模式與企業級應用

編輯：Java 程式設計入門

阻塞隊列核心概念

BlockingQueue 是支援兩種額外原子操作的隊列：

插入阻塞：隊列滿時，生產者執行緒阻塞直到有空位
移除阻塞：隊列空時，消費者執行緒阻塞直到有元素

自動平衡生產/消費速率，實現生產者-消費者模式的最佳工具。

四組核心 API 行為對比

操作	拋出異常	返回特殊值	一直阻塞	超時退出
插入	`add(e)`	`offer(e)`	`put(e)`	`offer(e, time, unit)`
移除	`remove()`	`poll()`	`take()`	`poll(time, unit)`
檢查	`element()`	`peek()`	-	-

行為詳解

拋出異常：違反隊列容量限制立即失敗
特殊值：offer 返回 boolean，poll/peek 返回 null
阻塞：put/take 無限等待，可響應中斷
超時：指定時間內嘗試，超時返回 false/null

JDK 七大阻塞隊列實現詳解

隊列	底層結構	邊界	鎖機制	特點	適用場景
`ArrayBlockingQueue`	環狀數組	有界	單一 ReentrantLock	公平/非公平可選，固定大小	固定緩衝區、高併發簡單場景
`LinkedBlockingQueue`	單向鏈表	有界(預設 Integer.MAX_VALUE)	雙鎖 (takeLock/putLock)	生產消費並行，吞吐高	高併發、大吞吐、無界風險需注意
`PriorityBlockingQueue`	優先隊列(堆)	無界	單一 ReentrantLock	元素需 Comparable/Comparator，僅阻塞 take	優先級任務、定時調度
`DelayQueue`	PriorityQueue	無界	ReentrantLock	元素實現 Delayed，到期才能取出	緩存過期、定時任務、重試延遲
`LinkedBlockingDeque`	雙向鏈表	有界	單一鎖	雙端操作，減少競爭	工作竊取、雙端緩衝
`LinkedTransferQueue`	鏈表	無界	CAS + 自旋	transfer() 直接移交，tryTransfer() 非阻塞嘗試	高併發移交、訊號傳遞
`SynchronousQueue`	無緩衝節點	無界(容量 0)	CAS + 雙棧/雙隊列	直接移交，無存儲，公平/非公平模式	執行緒池直接提交、高吞吐移交

核心實現深度解析

ArrayBlockingQueue：單鎖環狀數組

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull = lock.newCondition();
}

單一鎖控制 put/take，簡單但限制併發度
Condition 實現精確喚醒：notEmpty.await()、notFull.signal()
公平鎖：FIFO 順序獲取鎖，吞吐較低但避免飢餓

LinkedBlockingQueue：雙鎖鏈表

putLock 保護尾節點插入，takeLock 保護頭節點移除
生產/消費真正並行，高併發吞吐優於 ArrayBlockingQueue
無界預設值 Integer.MAX_VALUE，生產快於消費時 OOM 風險
Node 物件頻繁創建/回收，GC 壓力較大

SynchronousQueue：無緩衝直接移交

不存儲元素，生產者直接把元素移交給消費者
JDK 6+ 採用 雙棧/雙隊列 + CAS 無鎖算法，效能極高
公平模式：FIFO 隊列 + 公平鎖；非公平模式：LIFO 棧 + 非公平鎖（預設）
Executors.newCachedThreadPool() 內部使用

生產者-消費者實戰範例

檔案搜尋：一生產者 N 消費者

public class FileSearcher {
    public static void main(String[] args) {
        BlockingQueue<File> queue = new ArrayBlockingQueue<>(100);
        File rootDir = new File("/var/log");
        String keyword = "ERROR";
        int consumerCount = 10;
        
        // 生產者：遞歸枚舉檔案
        new Thread(() -> {
            try {
                enumerateFiles(rootDir, queue);
                // 投放毒丸結束信號
                for (int i = 0; i < consumerCount; i++) queue.put(POISON_PILL);
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }).start();
        
        // 消費者：搜尋關鍵字
        for (int i = 0; i < consumerCount; i++) {
            new Thread(() -> {
                try {
                    while (true) {
                        File file = queue.take();
                        if (file == POISON_PILL) {
                            queue.put(POISON_PILL); // 傳遞給下一個消費者
                            break;
                        }
                        searchFile(file, keyword);
                    }
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }).start();
        }
    }
    
    static final File POISON_PILL = new File("");
    
    static void enumerateFiles(File dir, BlockingQueue<File> queue) throws InterruptedException {
        File[] files = dir.listFiles();
        if (files == null) return;
        for (File f : files) {
            if (f.isDirectory()) enumerateFiles(f, queue);
            else queue.put(f);
        }
    }
    
    static void searchFile(File file, String keyword) {
        try (Scanner scanner = new Scanner(file)) {
            int lineNum = 0;
            while (scanner.hasNextLine()) {
                String line = scanner.nextLine();
                if (line.contains(keyword)) {
                    System.out.printf("%s:%d %s%n", file.getPath(), ++lineNum, line);
                }
            }
        } catch (IOException e) { /* 忽略 */ }
    }
}

DelayQueue 緩存過期實戰

public class CacheEntry implements Delayed {
    private final String key;
    private final long expireTime; // 納秒級絕對時間
    
    public CacheEntry(String key, long ttlMillis) {
        this.key = key;
        this.expireTime = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(ttlMillis);
    }
    
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(expireTime - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
    
    @Override
    public int compareTo(Delayed o) {
        long diff = this.expireTime - ((CacheEntry)o).expireTime;
        return (diff > 0) ? 1 : (diff < 0) ? -1 : 0;
    }
    
    public String getKey() { return key; }
}

// 清理執行緒
new Thread(() -> {
    while (!Thread.interrupted()) {
        try {
            CacheEntry entry = delayQueue.take(); // 阻塞直到過期
            cache.remove(entry.getKey());
            log.info("清理過期緩存: {}", entry.getKey());
        } catch (InterruptedException e) { break; }
    }
}).start();

效能對比與選型指南

場景	推薦隊列	理由
固定大小緩衝、簡單生產消費	ArrayBlockingQueue	預分配數組，無 GC 壓力，行為可預測
高併發、大吞吐、生產消費速率接近	LinkedBlockingQueue	雙鎖並行，吞吐最高，但需指定容量防 OOM
優先級任務、定時調度	PriorityBlockingQueue	堆排序，自然支援優先級
延遲執行、緩存過期、重試退避	DelayQueue	基於 PriorityQueue，到期自動可取
工作竊取、雙端操作	LinkedBlockingDeque	雙端減少競爭，ForkJoinPool 內部使用
高併發直接移交、執行緒池提交	SynchronousQueue	無緩衝，直接移交，吞吐極高
非阻塞嘗試移交、超時控制	LinkedTransferQueue	tryTransfer/transfer 靈活控制