java.util.concurrent.BlockingQueue 是構建高併發生產消費系統的基石。本文系統梳理 7 大核心實現類、四種操作語義、內部同步機制差異，並提供文件搜索、緩存淘汰、定時調度三大實戰場景完整代碼。
| 語義 | 插入方法 | 移除方法 | 行為特徵 |
|---|---|---|---|
| 拋出異常 | add(e) | remove() | 隊列滿/空時拋 IllegalStateException/NoSuchElementException |
| 返回特殊值 | offer(e) | poll() | 失敗返回 false/null，不阻塞 |
| 一直阻塞 | put(e) | take() | 直到成功或線程中斷，生產消費核心 API |
| 超時退出 | offer(e, t, u) | poll(t, u) | 超時返回 false/null，支援背壓控制 |
| 隊列 | 底層結構 | 邊界 | 鎖策略 | 適用場景 |
|---|---|---|---|---|
| ArrayBlockingQueue | 數組 | 有界 | 單鎖（可公平） | 固定緩衝區、高吞吐、公平調度 |
| LinkedBlockingQueue | 鏈表 | 有界（默認 Integer.MAX_VALUE） | 雙鎖（take/put 分離） | 高併發生產消費、默認首選 |
| PriorityBlockingQueue | 堆 | 無界 | 單公平鎖 | 優先級任務調度、不阻塞生產者 |
| DelayQueue | 堆 | 無界 | 單鎖 | 定時任務、緩存過期、訂單超時 |
| LinkedBlockingDeque | 雙向鏈表 | 有界 | 雙鎖 | 工作竊取、雙端操作 |
| LinkedTransferQueue | 鏈表 | 無界 | CAS+自旋 | 直接傳遞、低延遲交換 |
| SynchronousQueue | 無緩衝 | 0 容量 | CAS+雙棧/雙隊列 | 線程池直接提交、高吞吐交換 |
// 核心字段
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
// take() 只鎖 takeLock
public E take() throws InterruptedException {
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) notEmpty.await();
return dequeue();
} finally { takeLock.unlock(); }
}
// put() 只鎖 putLock
public void put(E e) throws InterruptedException {
final ReentrantLock putLock = this.putLock;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) notFull.await();
enqueue(e);
} finally { putLock.unlock(); }
}優勢： 生產者/消費者並行無競爭，高併發下吞吐量遠超單鎖 ArrayBlockingQueue。
風險： 無界模式下生產者過快導致 OOM，生產環境必須指定容量。
put 與 take 配對完成數據直接遞交Executors.newCachedThreadPool() 內部使用非公平模式public class FileKeywordSearch {
private static final int QUEUE_SIZE = 100;
private static final int SEARCH_THREADS = 50;
private static final File POISON = new File(""); // 終止信號
public static void main(String[] args) throws InterruptedException {
BlockingQueue queue = new ArrayBlockingQueue<>(QUEUE_SIZE);
// 生產者：遞歸枚舉文件
new Thread(() -> {
try { enumerate(new File("/var/log"), queue);
queue.put(POISON); }
catch (InterruptedException ignored) {}
}).start();
// 消費者：搜索關鍵字
for (int i = 0; i < SEARCH_THREADS; i++) {
new Thread(() -> {
try {
while (true) {
File file = queue.take();
if (file == POISON) { queue.put(POISON); break; }
search(file, "ERROR");
}
} catch (IOException | InterruptedException e) { Thread.currentThread().interrupt(); }
}).start();
}
}
static void enumerate(File dir, BlockingQueue queue) throws InterruptedException {
for (File f : dir.listFiles()) {
if (f.isDirectory()) enumerate(f, queue);
else queue.put(f);
}
}
static void search(File file, String keyword) throws IOException { ... }
} 關鍵技巧： 使用 POISON 對象傳遞終止信號，消費者收到後重新放回以通知其他消費者。
public class CacheEntry implements Delayed {
private final String key;
private final long expireTime; // 納秒時間戳
public CacheEntry(String key, long ttlMs) {
this.key = key;
this.expireTime = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(ttlMs);
}
@Override public long getDelay(TimeUnit unit) {
return unit.convert(expireTime - System.nanoTime(), TimeUnit.NANOSECONDS);
}
@Override public int compareTo(Delayed o) {
return Long.compare(this.expireTime, ((CacheEntry)o).expireTime);
}
}
// 淘汰線程
new Thread(() -> {
while (!Thread.interrupted()) {
try {
CacheEntry expired = delayQueue.take(); // 阻塞直到最近條目過期
cacheMap.remove(expired.key);
} catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
}).start();優勢： 無需定時掃描，take() 精準阻塞到最近過期時間，CPU 占用極低。
// 核心線程數 0，最大無界，隊列 SynchronousQueue
ThreadPoolExecutor executor = new ThreadPoolExecutor(
0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
new SynchronousQueue<>(),
new ThreadFactoryBuilder().setNameFormat("worker-%d").build(),
new ThreadPoolExecutor.AbortPolicy()
);
// 任務提交即創建線程，空閒 60s 回收，適合突發型高併發需要優先級？ → PriorityBlockingQueue 需要延時執行？ → DelayQueue 需要雙端操作/工作竊取？ → LinkedBlockingDeque 追求極致吞吐/直接交換？ → SynchronousQueue / LinkedTransferQueue 通用高併發生產消費？ → LinkedBlockingQueue (指定容量!) 固定緩衝/需公平調度？ → ArrayBlockingQueue (公平模式)
阻塞隊列不只是容器，更是並發協調協議。理解每種隊列的鎖策略、內存模型、背壓機制，才能在高併發系統中做出正確架構決策。
