JDK源碼分析（11）之 BlockingQueue 相關

本文將主要結合源碼對 JDK 中的阻塞隊列進行分析，并比較其各自的特點；

一、BlockingQueue 概述

說到阻塞隊列想到的第一個應用場景可能就是生產者消費者模式了，如圖所示；

根據上圖所示，明顯在入隊和出隊的時候，會發生競爭；所以一種很自然的想法就是使用鎖，而在 JDK 中也的確是通過鎖來實現的；所以?BlockingQueue?的源碼其實可以當成鎖的應用示例來查看；同時 JDK 也為我們提供了多種不同功能的隊列：

• ArrayBlockingQueue?：基于數組的有界隊列；

• LinkedBlockingQueue?：基于鏈表的×××隊列（可以設置容量）；

• PriorityBlockingQueue?：基于二叉堆的×××優先級隊列；

• DelayQueue?：基于 PriorityBlockingQueue 的×××延遲隊列；

• SynchronousQueue?：無容量的阻塞隊列（Executors.newCachedThreadPool() 中使用的隊列）；

• LinkedTransferQueue?：基于鏈表的×××隊列；

接下來我們就對最常用的?ArrayBlockingQueue?和?LinkedBlockingQueue?進行分析；

二、 ArrayBlockingQueue 源碼分析

1. 結構概述

public?class?ArrayBlockingQueue<E>?extends?AbstractQueue<E>?implements?BlockingQueue<E>,?java.io.Serializable?{????final?Object[]?items;???????????????//?容器數組
????int?takeIndex;??????????????????????//?出隊索引
????int?putIndex;???????????????????????//?入隊索引
????int?count;??????????????????????????//?排隊個數
????final?ReentrantLock?lock;???????????//?全局鎖
????private?final?Condition?notEmpty;???//?出隊條件隊列
????private?final?Condition?notFull;????//?入隊條件隊列
????...
}

ArrayBlockingQueue?的結構如圖所示：

如圖所示，

• ArrayBlockingQueue?的數組其實是一個邏輯上的環狀結構，在添加、取出數據的時候，并沒有像?ArrayList?一樣發生數組元素的移動（當然除了?removeAt(final int removeIndex)）；

• 并且由?takeIndex?和?putIndex?指示讀寫位置；

• 在讀寫的時候還有兩個讀寫條件隊列；

下面我們就讀寫操作，對源碼簡單分析：

2. 入隊

public?void?put(E?e)?throws?InterruptedException?{
??checkNotNull(e);??final?ReentrantLock?lock?=?this.lock;
??lock.lockInterruptibly();??try?{????while?(count?==?items.length)??//?當隊列已滿的時候放入?putCondition?條件隊列
??????notFull.await();???
????enqueue(e);??//?入隊
??}?finally?{
????lock.unlock();
??}
}

private?void?enqueue(E?x)?{??//?assert?lock.getHoldCount()?==?1;
??//?assert?items[putIndex]?==?null;
??final?Object[]?items?=?this.items;
??items[putIndex]?=?x;??//?插入隊列
??if?(++putIndex?==?items.length)?putIndex?=?0;??//?指針走一圈的時候復位
??count++;
??notEmpty.signal();??//?喚醒?takeCondition?條件隊列中等待的線程}

3. 出隊

public?E?take()?throws?InterruptedException?{??final?ReentrantLock?lock?=?this.lock;
??lock.lockInterruptibly();??try?{????while?(count?==?0)??//?當隊列為空的時候，放入?takeCondition?條件
??????notEmpty.await();??
????return?dequeue();???//?出隊
??}?finally?{
????lock.unlock();
??}
}

private?E?dequeue()?{??//?assert?lock.getHoldCount()?==?1;
??//?assert?items[takeIndex]?!=?null;
??final?Object[]?items?=?this.items;??@SuppressWarnings("unchecked")
??E?x?=?(E)?items[takeIndex];??//?取出元素
??items[takeIndex]?=?null;??if?(++takeIndex?==?items.length)
????takeIndex?=?0;
??count--;??if?(itrs?!=?null)
????itrs.elementDequeued();
??notFull.signal();??//?取出元素后，隊列空出一位，所以喚醒?putCondition?中的線程
??return?x;
}

三、LinkedBlockingQueue 源碼分析

1. 結構概述

public?class?LinkedBlockingQueue<E>?extends?AbstractQueue<E>?implements?BlockingQueue<E>,?java.io.Serializable?{??
??private?final?int?capacity;?//?默認?Integer.MAX_VALUE
??private?final?AtomicInteger?count?=?new?AtomicInteger();?//?容量
??transient?Node<E>?head;??????????//?頭結點?head.item?==?null
??private?transient?Node<E>?last;??//?尾節點?last.next?==?null
??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;?}
??}
}

LinkedBlockingQueue?的結構如圖所示：

如圖所示，

• LinkedBlockingQueue?其實就是一個簡單的單向鏈表，其中頭部元素的數據為空，尾部元素的 next 為空；

• 因為讀寫都有競爭，所以在頭部和尾部分別有一把鎖；同時還有對應的兩個條件隊列；

下面我們就讀寫操作，對源碼簡單分析：

2. 入隊

public?boolean?offer(E?e)?{??if?(e?==?null)?throw?new?NullPointerException();??final?AtomicInteger?count?=?this.count;??if?(count.get()?==?capacity)?return?false;??//?如果隊列已滿，直接返回失敗
??int?c?=?-1;
??Node<E>?node?=?new?Node<E>(e);??????????????//?將數據封裝為節點
??final?ReentrantLock?putLock?=?this.putLock;
??putLock.lock();??try?{????if?(count.get()?<?capacity)?{
??????enqueue(node);??????????????????????????//?入隊
??????c?=?count.getAndIncrement();??????if?(c?+?1?<?capacity)???????????????????//?如果隊列未滿，則繼續喚醒?putCondition?條件隊列
????????notFull.signal();
????}
??}?finally?{
????putLock.unlock();
??}??if?(c?==?0)???????????//?如果添加之前的容量為0，說明在出隊的時候有競爭，則喚醒?takeCondition
????signalNotEmpty();???//?因為是兩把鎖，所以在喚醒?takeCondition的時候，還需要獲取?takeLock
??return?c?>=?0;
}

private?void?enqueue(Node<E>?node)?{??//?assert?putLock.isHeldByCurrentThread();
??//?assert?last.next?==?null;
??last?=?last.next?=?node;??//?連接節點，并設置尾節點}

3. 出隊

public?E?take()?throws?InterruptedException?{
??E?x;??int?c?=?-1;??final?AtomicInteger?count?=?this.count;??final?ReentrantLock?takeLock?=?this.takeLock;
??takeLock.lockInterruptibly();??try?{????while?(count.get()?==?0)?{???//?如果隊列為空，則加入?takeCondition?條件隊列
??????notEmpty.await();
????}
????x?=?dequeue();???????????????//?出隊
????c?=?count.getAndDecrement();????if?(c?>?1)
??????notEmpty.signal();?????????//?如果隊列還有剩余，則繼續喚醒?takeCondition?條件隊列
??}?finally?{
????takeLock.unlock();
??}??if?(c?==?capacity)?????????????//?如果取之前隊列是滿的，說明入隊的時候有競爭，則喚醒?putCondition
????signalNotFull();?????????????//?同樣注意是兩把鎖
??return?x;
}

private?E?dequeue()?{??//?assert?takeLock.isHeldByCurrentThread();
??//?assert?head.item?==?null;
??Node<E>?h?=?head;
??Node<E>?first?=?h.next;
??h.next?=?h;?//?help?GC???//?將next引用指向自己，則該節點不可達，在下一次GC的時候回收
??head?=?first;
??E?x?=?first.item;
??first.item?=?null;??return?x;
}

四、ABQ、LBQ 對比

根據以上的講解，我們可以逐步分析出一些不同，以及在不同場景隊列的選擇：

1. 結構不同

• ABQ：基于數組，有界，一把鎖；

• LBQ：基于鏈表，×××，兩把鎖；

內存分配

• ABQ：隊列空間預先初始化，受堆空間影響小，穩定性高；

• LBQ：隊列空間動態變化，受對空間影響大，穩定性差；

入隊、出隊效率

• ABQ：數據直接賦值，移除；隊列空間重復使用，效率高；

• LBQ：數據需要包裝為節點；需開辟新空間，效率低；

競爭方面

• ABQ：出入隊共用一把鎖，相互影響；競爭嚴重時效率低；

• LBQ：出入隊分用兩把鎖，互不影響；競爭嚴重時效率影響??；

所以在這里并不能簡單的給出詳細的數據，證明哪個隊列更適合什么場景，最好是結合實際使用場景分析。