Java 併發隊列深度對比：LinkedBlockingQueue vs ArrayBlockingQueue

前言

BlockingQueue 是 Java 併發包中核心的阻塞隊列介面，LinkedBlockingQueue 和 ArrayBlockingQueue 是最常用的兩種實作。雖然都實作了阻塞隊列語意，但底層資料結構、鎖機制、效能特徵差異巨大。本文從原始碼、效能、適用場景三個維度深度對比。

核心差異總覽

原始碼級深度解析

1. LinkedBlockingQueue：雙鎖 + 鏈表

// 核心欄位
private final E[] items;           // ArrayBlockingQueue 使用陣列
private final ReentrantLock lock;  // 單一鎖

// LinkedBlockingQueue 核心欄位
private final int capacity;        // 容量上限
private final AtomicInteger count = new AtomicInteger(0); // 原子計數器

// 雙鎖設計
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();

// 節點類別
static class Node<E> {
    E item;
    Node<E> next;
    Node(E x) { item = x; }
}

// 隊列頭尾指針
private transient Node<E> head;
private transient Node<E> last;

入隊：

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    Node<E> node = new Node<>(e);  // 每次新建 Node 物件
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            notFull.await();        // 滿時等待
        }
        enqueue(node);              // 尾插法
        if (count.getAndIncrement() == 0) {
            notEmpty.signal();      // 通知消費者
        }
    } finally {
        putLock.unlock();
    }
}

private void enqueue(Node<E> node) {
    last = last.next = node;
}

出隊：

public E take() throws InterruptedException {
    final ReentrantLock takeLock = this.takeLock;
    final AtomicInteger count = this.count;
    takeLock.lockInterruptibly();
    try {
        while (count.get() == 0) {
            notEmpty.await();       // 空時等待
        }
        E x = dequeue();            // 頭刪法
        if (count.getAndDecrement() == capacity) {
            notFull.signal();       // 通知生產者
        }
        return x;
    } finally {
        takeLock.unlock();
    }
}

private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

2. ArrayBlockingQueue：單鎖 + 循環陣列

// 核心欄位
final Object[] items;              // 固定大小陣列
int takeIndex;                     // 取元素索引
int putIndex;                      // 放元素索引
int count;                         // 當前元素數

final ReentrantLock lock;
final Condition notEmpty;
final Condition notFull;

入隊/出隊：

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length) {
            notFull.await();
        }
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

private void enqueue(E x) {
    items[putIndex] = x;
    if (++putIndex == items.length) putIndex = 0; // 循環
    count++;
    notEmpty.signal();
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0) {
            notEmpty.await();
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null; // help GC
    if (++takeIndex == items.length) takeIndex = 0;
    count--;
    notFull.signal();
    return x;
}

效能基準測試

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Group)
public class BlockingQueueBenchmark {

    @Param({"100", "1000", "10000"})
    int capacity;

    @Param({"1", "4", "16"})
    int threads;

    LinkedBlockingQueue<Integer> linkedQueue;
    ArrayBlockingQueue<Integer> arrayQueue;

    @Setup
    public void setup() {
        linkedQueue = new LinkedBlockingQueue<>(capacity);
        arrayQueue = new ArrayBlockingQueue<>(capacity);
    }

    @Benchmark
    @Group("linked")
    @GroupThreads(4)
    public void linkedPut(LinkedState state) throws InterruptedException {
        linkedQueue.put(state.value);
    }

    @Benchmark
    @Group("linked")
    @GroupThreads(4)
    public void linkedTake(LinkedState state) throws InterruptedException {
        state.value = linkedQueue.take();
    }

    @Benchmark
    @Group("array")
    @GroupThreads(4)
    public void arrayPut(ArrayState state) throws InterruptedException {
        arrayQueue.put(state.value);
    }

    @Benchmark
    @Group("array")
    @GroupThreads(4)
    public void arrayTake(ArrayState state) throws InterruptedException {
        state.value = arrayQueue.take();
    }
}

典型測試結果（參考值，視硬體而定）

記憶體與 GC 影響分析

// LinkedBlockingQueue: 每次 put 分配
Node<E> node = new Node<>(e);  // ~24 bytes + object header
// 高頻場景：每秒百萬次 → 每秒 ~24MB 分配 → Young GC 壓力大

// ArrayBlockingQueue: 零分配
items[putIndex] = x;  // 直接寫入預分配陣列
// 無額外物件分配，GC 幾乎無影響

GC 壓力對比

進階特性差異

1. clear() 方法

// LinkedBlockingQueue: 需獲取雙鎖
public void clear() {
    takeLock.lock();
    try {
        putLock.lock();
        try {
            // 清空鏈表
            for (Node<E> p = head.next; p != null; p = p.next) {
                p.item = null;
            }
            head.last = head;
            count.set(0);
        } finally {
            putLock.unlock();
        }
    } finally {
        takeLock.unlock();
    }
}

// ArrayBlockingQueue: 單鎖，簡單高效
public void clear() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        for (int i = takeIndex; i != putIndex; i = (i + 1) % items.length) {
            items[i] = null;
        }
        putIndex = takeIndex = 0;
        count = 0;
    } finally {
        lock.unlock();
    }
}

2. size() 實作

// LinkedBlockingQueue: AtomicInteger 原子讀取
public int size() { return count.get(); }

// ArrayBlockingQueue: 鎖保護下的 volatile int
public int size() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try { return count; } finally { lock.unlock(); }
}
// 注意：ArrayBlockingQueue 的 size() 有鎖開銷，高頻調用影響效能

3. 迭代器弱一致性

• 兩者皆為弱一致迭代器：反映迭代器建立時刻的狀態，不拋出 ConcurrentModificationException
• LinkedBlockingQueue：遍歷鏈表快照
• ArrayBlockingQueue：遍歷陣列快照

選型決策樹

                    需要無界隊列？
                         │
            ┌────────────┴────────────┐
            ▼                         ▼
           是                          否
            │                         │
            ▼                         ▼
    LinkedBlockingQueue      固定容量已知？
    (預設 Integer.MAX_VALUE)      │
            │            ┌────────┴────────┐
            ▼            ▼                 ▼
        直接使用       是                 否
                    ▼                 ▼
              ArrayBlockingQueue   預估峰值？
                    │            ┌──────┴──────┐
                    ▼            ▼             ▼
               直接使用      可估算        難以估算
                                 │            │
                                 ▼            ▼
                            ArrayBlockingQueue  LinkedBlockingQueue
                            (指定容量)          (設上限)

實戰場景建議

常見誤區

誤區 1：LinkedBlockingQueue 預設無界，可以無限放

危險！生產者快於消費者會導致 OOM。生產環境必須指定容量上限。

誤區 2：ArrayBlockingQueue 效能一定較差

單鎖在低爭用時確實較慢，但批量操作、記憶體敏感場景反而更優。不要只看微基準測試。

誤區 3：size() 是即時準確的

弱一致性：返回值僅供參考，不可用於精確控制邏輯。

誤區 4：雙鎖完全無競爭

takeLock/putLock 仍會在 count 變量、head/last 指針更新時產生緩存一致性流量。

總結

LinkedBlockingQueue 以雙鎖 + 鏈表換取高併發吞吐，適合生產消費速度相近、需要動態緩衝的場景；ArrayBlockingQueue 以單鎖 + 循環陣列換取零分配、確定性記憶體，適合延遲敏感、記憶體受限、容量固定的場景。選型核心在於：吞吐優先選 Linked，延遲/記憶體優先選 Array。