多執行緒程式設計-分析阻塞佇列的原始碼實現
阿新 • • 發佈：2020-07-30
> 本文在個人技術部落格同步釋出，詳情可[**用力戳**](http://www.17coding.info/article/28)
> 亦可掃描螢幕右側二維碼關注個人公眾號，公眾號內有個人聯絡方式，等你來撩...
看過我[上一篇](http://www.17coding.info/article/27)文章的應該知道（家裡條件允許的可以先看看上一篇文章），如果想實現一個生產者消費者模型，我們可以基於JVM自帶的synchronized+wait+notify實現，也可以用JDK裡面的ReentrantLock+Condition實現！不過從上篇文章的demo看，實現起來也不是那麼容易！因為我們既要關心什麼時候需要阻塞執行緒，又要需要關心何時喚醒執行緒。控制的細節太多，一個疏忽可能就導致了一個不易發現的bug，比如上篇文章中的虛假喚醒的例子！那有沒有一種我們不用關心那麼多複雜細節就能實現生產者消費者模式的方法呢？本文要講的阻塞佇列就是一種很好的實現！
在我們剛開始學資料結構的時候，都接觸過一種先進先出（first in first out，簡稱“FIFO”）的資料結構，叫佇列。阻塞佇列從名字看也是佇列的一種，因此滿足佇列的特性，然後這個佇列是可阻塞的！這個阻塞怎麼理解呢？就是當我們一個執行緒往阻塞佇列裡面新增元素的時候，如果佇列滿了，那這個執行緒不會直接返回，而是會被阻塞，直到元素新增成功！當我們一個執行緒從阻塞佇列裡面獲取元素的時候，如果佇列是空的，那這個執行緒不會直接返回，而是會被阻塞直到元素獲取成功。而阻塞以及喚醒的操作都由阻塞佇列來管理！
## 常用阻塞佇列類圖
我們先看在java中阻塞佇列基本的繼承關係圖：
完整的繼承關係要比這張圖複雜一些，但為了清晰起見圖中我只畫了主要的類和關係。佇列的基介面Queue與我們開發中經常用到的List、Set是兄弟關係，因此我這裡也列出來了方便對比記憶！阻塞佇列的基介面是繼承自Queue介面的BlockingQueue介面，其他阻塞佇列具體實現都繼承BlockingQueue介面！
## BlockingQueue常用方法
我們先看佇列基介面Queue中的方法
這個介面一共6個方法，我們可以分為兩組
**1、“異常”組**
> 1、add(e)：將元素放到佇列末尾，成功返回true，失敗則拋異常。
> 2、remove()：獲取並移除隊首元素，獲取失敗則拋異常。
> 3、element()：獲取隊首元素，不移除，獲取失敗則拋異常。
**2、“特殊值”組**
> 1、offer(e)：將元素放到佇列末尾，成功返回true，失敗返回false。
> 2、poll()：獲取並返回隊首元素，獲取失敗則返回null。
> 3、peek()：獲取隊首元素，不移除，獲取失敗則返回null。
“異常”組的3個方法在操作失敗的時候會拋異常，因此叫“異常”組！
“特殊值”組3個方法與“異常”組的3個方法是一一對應的，功能都一樣，只是在操作失敗的時候不會拋異常而是返回一個特殊值，因此叫“特殊值組”。
這兩組方法都是在Queue介面中定義的，因此跟阻塞就沒有什麼關係了。那我們再看看BlockingQueue介面中的方法
這個介面我們重點關注標記出來的4個方法，這幾個方法我們也可以分為兩組
**3、“阻塞”組**
> 1、put(e)：將元素放到佇列末尾，如果佇列滿了，則等待。
> 2、take()：獲取並移除隊首元素，如果佇列為空，則等待。
** 4、“超時”組**
> 1、offer(e,time,unit)：將元素放到佇列末尾，如果佇列滿了，則等待，當等待超過指定時間後仍新增元素失敗，則返回false，否則返回true。
> 2、poll(time,unit)：獲取並返回隊首元素，如果佇列為空，則等待，當等待超過指定時間後仍獲取失敗則返回null，否則返回獲取到的元素。
這兩組方法都是在BlockingQueue介面中定義的，因此都是跟阻塞相關的！
“阻塞”組2個方法在操作不成功的時候會一直阻塞執行緒，直到能夠操作成功，因此叫“阻塞”組！用一個成語形容就是“**不見不散**”！
“超時”組2個方法與“超時”組的2個方法是一一對應的，功能都一樣，只是這2個方法不會一直阻塞，超過了指定的時間還沒成功就停止阻塞並返回，因此叫“超時”組！用一個成語形容就是“**過時不候**”！
這四組方法合在一起就有了下面的一張表格：
| 方法功能 | 異常組 | 特殊值組 | 阻塞組 | 超時組 |
| -------- | --------- | -------- | ------ | ------------------ |
| 元素入隊 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 元素出隊 | remove() | pool() | take() | poll(time,unit) |
| 檢查元素 | element() | peek() | 無 | 無 |
## 原始碼分析常用阻塞佇列
BlockingQueue的實現類有多個，但是如果每一個原始碼都進行分析那不僅很影響篇幅且沒必要，因此我這裡拿三個常用的阻塞佇列原始碼進行分析！在原始碼中jdk的版本為1.8！
### ArrayBlockingQueue
我們先看下ArrayBlockingQueue中的幾個屬性
```java
/** The queued items 使用陣列儲存元素 */
final Object[] items;
/** items index for next take, poll, peek or remove 下一個出隊元素索引 */
int takeIndex;
/** items index for next put, offer, or add 下一個入隊元素索引 */
int putIndex;
/** Number of elements in the queue 佇列元素個數 */
int count;
/*
* ReentrantLock+Condition控制併發
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
```
> 1.object型別陣列，也意味著ArrayBlockingQueue底層資料結構是陣列。
> 2.ReentrantLock+Condition，如果看過我上一篇文章的應該很熟悉，這是用做來執行緒同步和執行緒通訊的。
我們再看下ArrayBlockingQueue的建構函式。
```java
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
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();
}
public ArrayBlockingQueue(int capacity, boolean fair,
Collection c){
this(capacity, fair);
//初始化一個集合到佇列
....
}
```
這三個建構函式都必須傳入一個int型別的capacity引數，這個引數也意味著ArrayBlockingQueue是一個有界的阻塞佇列！
我們前面說過佇列有常用的四組方法，而跟阻塞相關的是“阻塞”組和“超時”組的四個方法！我們以“阻塞”組的put()和take()方法為例，來窺探一下原始碼裡面的奧祕：
```java
/**
* Inserts the specified element at the tail of this queue, waiting
* for space to become available if the queue is full.
*/
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) {
final Object[] items = this.items;
items[putIndex] = x;
// 下一個入隊元素索引超過了陣列的長度，則又從0開始。
if (++putIndex == items.length)
putIndex = 0;
count++;
//放入元素後，釋放一個“不空”的訊號。喚醒等待中的出隊執行緒。
notEmpty.signal();
}
```
```java
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;
E x = (E) items[takeIndex];
items[takeIndex] = null;//help GC
// 下一個出隊元素索引超過了陣列的長度，則又從0開始。
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();//更新迭代器元素資料
//取出元素後，釋放一個“不滿”的訊號。喚醒等待中的入隊執行緒。
notFull.signal();
return x;
}
```
ArrayBlockingQueue的入隊出隊程式碼還是很簡單的，當我們往一個阻塞佇列裡面新增資料的時候，阻塞佇列用一個固定長度的資料儲存資料，如果陣列的長度達到了最大容量，則新增資料的執行緒會被阻塞。當我們從阻塞佇列獲取資料的時候，如果佇列為空，則獲取資料的執行緒會被阻塞！相信程式碼上的註釋已經足夠理解這塊的程式碼邏輯了！
### LinkedBlockingQueue
我們先看下LinkedBlockingQueue中的幾個屬性
```java
/** The capacity bound, or Integer.MAX_VALUE if none 佇列容量 */
private final int capacity;
/** Current number of elements 佇列元素個數 */
private final AtomicInteger count = new AtomicInteger();
/**
* 佇列頭
* Head of linked list.
* Invariant: head.item == null
*/
transient Node head;
/**
* 佇列尾
* Tail of linked list.
* Invariant: last.next == null
*/
private transient Node last;
/** Lock held by take, poll, etc 出隊操作用到的鎖 */
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc 入隊操作用到的鎖 */
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();
```
> 1.Node型別的變數head和last，這是連結串列常見操作，也意味著LinkedBlockingQueue底層資料結構是連結串列。
> 2.與ArrayBlockingQueue不同的是，這裡有兩個ReentrantLock物件，put操作個take操作的鎖物件是分開的，這樣做也是為了提高容器的併發能力。
再看下Node這個內部類
```java
/**
* Linked list node class
*/
static class Node {
E item;
//指向下一個節點
Node next;
Node(E x) { item = x; }
}
```
只有next屬性意味著這是一個單向連結串列！
再看下LinkedBlockingQueue的建構函式
```java
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node(null);
}
public LinkedBlockingQueue(Collection c) {
this(Integer.MAX_VALUE);
...
}
```
> 1.當建構函式不傳capacity引數的時候，LinkedBlockingQueue就是一個無界阻塞佇列（其實也並非無界，不傳預設值就是Integer.MAX_VALUE）。
> 2.當建構函式傳入capacity引數的時候，LinkedBlockingQueue就是一個有界阻塞佇列。
我們依然看看在LinkedBlockingQueue中“阻塞”組的兩個方法put()和take()分別怎麼實現的
```java
/**
* Inserts the specified element at the tail of this queue, waiting if
* necessary for space to become available.
*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
//儲存佇列元素數量
int c = -1;
//建立新節點
Node node = new Node(e);
//獲取putLock
final ReentrantLock putLock = this.putLock;
//佇列元素數量
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
//判斷佇列是否滿足入隊條件，如果佇列已滿，則阻塞等待一個“不滿”的訊號
while (count.get() == capacity) {
notFull.await();
}
//入隊操作
enqueue(node);
//佇列元素數量+1，執行完下面這句後，count是入隊後的元素數量，而c的值還是入隊前的元素數量。
c = count.getAndIncrement();
//當前入隊操作成功後，如果元素數量還小於佇列容量，則釋放一個“不滿”的訊號
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
//這裡的c前面說了是元素入隊前的數量，如果入隊前元素數量為0（佇列是空的），那可能會有出隊執行緒在等待一個“不空”的訊號，所以這裡釋放一個“不空”的訊號。
if (c == 0)
signalNotEmpty();
}
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
public E take() throws InterruptedException {
//出隊元素
E x;
//儲存佇列元素數量
int c = -1;
//佇列元素數量
final AtomicInteger count = this.count;
//獲取takeLock
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
//判斷佇列是否滿足出隊條件，如果佇列為空，則阻塞等待一個“不空”的訊號
while (count.get() == 0) {
notEmpty.await();
}
//出隊操作
x = dequeue();
//佇列元素數量-1，執行完下面這句後，count是出隊後的元素數量，而c的值還是出隊前的元素數量。
c = count.getAndDecrement();
//當前出隊操作成功前佇列元素大於1，那當前出隊操作成功後佇列元素也就大於0，則釋放一個“不空”的訊號
if (c > () : new TransferStack();
}
```
兩個建構函式，fair引數指定公平策略，預設為false，因此是非公平模式！先看看put和take方法的實現：
```java
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
if (transferer.transfer(e, false, 0) == null) {
Thread.interrupted();
throw new InterruptedException();
}
}
public E take() throws InterruptedException {
E e = transferer.transfer(null, false, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
```
put和take方法很類似，都是呼叫transferer.transfer(...)方法，區別在於第一個引數！put方法在呼叫時候會參入入隊的值，而take方法傳入null。
上面說過有公平和非公平策略，今天將重點分析公平模式TransferQueue的原始碼！從名字能看出來這也是一個佇列，我們先看TransferQueue的重點屬性和構造方法：
```java
// 指向佇列頭部
transient volatile QNode head;
// 指向佇列尾部
transient volatile QNode tail;
TransferQueue() {
//初始化一個空
//#1
QNode h = new QNode(null, false); // initialize to dummy node.
head = h;
tail = h;
}
```
一頭一尾，連結串列的一貫操作！構造方法中，建立了一個QNode結點，並且將head和tail都指向這個結點！我們再看看QNode類的重要屬性和構造方法：
```java
volatile QNode next; // 指向佇列的下一個節點
volatile Object item; // 節點儲存的元素
volatile Thread waiter; // 被阻塞的執行緒
final boolean isData; // 是否是“資料”結點（入隊執行緒為true，出隊執行緒為false）
QNode(Object item, boolean isData) {
this.item = item;
this.isData = isData;
}
```
我們再回到上面提到的transferer.transfer(...)方法，也就是TransferQueue中的transfer(...)方法，核心邏輯都在這個方法中體現：
```java
/**
* “存”或者“取”一個元素
*/
@SuppressWarnings("unchecked")
E transfer(E e, boolean timed, long nanos) {
QNode s = null; // constructed/reused as needed
//當前操作型別，傳非null的值則為生產執行緒，傳null則為消費執行緒。
boolean isData = (e != null);
for (;;) {
QNode t = tail;
QNode h = head;
//上面我們說過在構造方法中就建立了一個QNode結點，並且將head和tail都指向這個結點
//因此這裡t、h一般情況下不會為null
if (t == null || h == null) // saw uninitialized value
continue; // spin
//根據SynchronousQueue的特性，不同型別的操作會配對成功。
//因此在阻塞佇列中只會存在一種型別的阻塞節點，要麼全是消費執行緒要麼全是生產執行緒！
//所以分三種情況：
//1.h == t，這種情況下佇列為空，需要將當前節點入隊。
//2.t.isData == isData尾部節點的操作型別與當前操作型別
// 一致(尾部節點的操作型別代表著佇列中所有節點的操作型別)，需要將當前節點入隊。
//3.佇列不為空且尾部節點的操作型別與當前操作型別不一致，
// 需要從佇列頭部匹配一個節點並返回。
//因此再看下面的程式碼，會根據上面3種情況走不同的分支。
if (h == t || t.isData == isData) { // empty or same-mode
//進入這個分支就是上面1、2的情況
//獲取尾部節點的next指向，正常情況下tn等於null
QNode tn = t.next;
//下面是判斷是否出現併發導致尾節點被更改
if (t != tail) // inconsistent read
continue;
if (tn != null) { // lagging tail
advanceTail(t, tn);
continue;
}
//超時判斷
if (timed && nanos <= 0) // can't wait
return null;
//將當前操作建立為新節點，傳入資料值和操作型別。
//#2
if (s == null)
s = new QNode(e, isData);
//1、將阻塞佇列中尾部節點的next指向新節點
//2、將tail屬性的指向設定為新節點
//#3
if (!t.casNext(null, s)) // failed to link in
continue;
advanceTail(t, s); // swing tail and wait
//在這個方法內部會進行自旋或者阻塞，直到配對成功。
//建議這裡先跳到下面這個方法內部看完邏輯再回來。
Object x = awaitFulfill(s, e, timed, nanos);
//只有在執行緒被中斷的情況下會進入這個分支
if (x == s) { // wait was cancelled
clean(t, s);
return null;
}
if (!s.isOffList()) { // not already unlinked
advanceHead(t, s); // unlink if head
if (x != null) // and forget fields
s.item = s;
s.waiter = null;
}
//如果為生產執行緒，則返回入隊的值；如果為消費執行緒，則返回匹配到的生產執行緒的值。
return (x != null) ? (E)x : e;
} else { // complementary-mode
//進入這個分支就是上面3的情況
//找到頭部節點的next指向
//#4
QNode m = h.next; // node to fulfill
if (t != tail || m == null || h != head)
continue; // inconsistent read
Object x = m.item;
//m.casItem(x, e)方法很重要，會將匹配到的節點的item修改為當前操作的值。
//這樣awaitFulfill方法的x != e條件才能成立，被匹配的阻塞執行緒才能返回。
//#5
if (isData == (x != null) || // m already fulfilled
x == m || // m cancelled
!m.casItem(x, e)) { // lost CAS
advanceHead(h, m); // dequeue and retry
continue;
}
//調整head屬性的指向，這裡建議這裡先跳到下面這個方法內部看完邏輯再回來。
advanceHead(h, m); // successfully fulfilled
//喚醒匹配到的阻塞執行緒
LockSupport.unpark(m.waiter);
//如果為生產執行緒，則返回入隊的值；如果為消費執行緒，則返回匹配到的生產執行緒的值。
return (x != null) ? (E)x : e;
}
}
}
Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
/* Same idea as TransferStack.awaitFulfill */
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
//如果頭節點的next指向當前的資料節點，也就是當前資料節點是下一個待匹配的節點，那就自旋等待一會兒。
//如果設定了超時時間就少自旋一會兒，沒有設定超時時間就多自旋一會兒。
//可以看看maxTimedSpins和maxUntimedSpins兩個屬性的值設定，是與cpu數量相關的。
int spins = ((head.next == s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())
s.tryCancel(e);
Object x = s.item;
// 第一次進來這裡肯定是相等的，所以不會進入這個分支。
// 當有其他的執行緒匹配到當前節點，這裡的s.item的值會被更改（前面說到過的m.casItem(x, e)方法），所以方法返回。
if (x != e)
return x;
if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
s.tryCancel(e);
continue;
}
}
if (spins > 0)
--spins;
else if (s.waiter == null)
s.waiter = w;
else if (!timed)
//這裡執行緒會阻塞，如果有執行緒與當前執行緒匹配，則被喚醒進行下一次迴圈。
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}
void advanceHead(QNode h, QNode nh) {
//這個方法做了兩個操作
//1、將head屬性的指向調整為頭節點的下一個結點
//2、將原頭節點的next指向原頭節點本身
//#6
if (h == head &&
UNSAFE.compareAndSwapObject(this, headOffset, h, nh))
h.next = h; // forget old next
}
```
不知道看完上面的SynchronousQueue基於公平模式TransferQueue的原始碼有沒有對SynchronousQueue有一個很好的瞭解！下面我模擬了一個場景，先有一個生產執行緒進入佇列，然後一個消費執行緒進入佇列。結合上面原始碼我畫了幾張節點變化的圖例以便更好的理解上面整個過程，可以結合上面的原始碼一起看
```java
//建立SynchronousQueue物件
SynchronousQueue synchronousQueue = new SynchronousQueue<>(true);
//生產執行緒
new Thread(new Runnable() {
@Override
public void run() {
try {
synchronousQueue.put("VALUE");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
Thread.sleep(1000);
//消費執行緒
new Thread(new Runnable() {
@Override
public void run() {
try {
synchronousQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
```
我們在建立SynchronousQueue物件時候會執行建構函式，也就是在原始碼#1處執行完後，會建立一個新的節點node,如下圖所示，一頭一尾都指向建構函式中創建出來的新節點node！
然後會執行synchronousQueue.put()的邏輯，也就是TransferQueue中的transfer(...)方法邏輯。按照我們之前的分析，會執行到原始碼#2處，執行完後新的節點node1會被建立，如下圖所示。
接著在程式碼#3處執行完後，節點圖示如下，注意紅色箭頭指向的調整。
到這裡，生產執行緒會進入awaitFulfill方法自旋後阻塞！等待消費執行緒的喚醒！
然後執行synchronousQueue.take()的邏輯，也就是TransferQueue中的transfer(...)方法邏輯。按照我們之前的分析，會執行到原始碼#4處，執行完後就找到了我們需要匹配的節點node1，注意紅色箭頭指向。
執行到#5處的方法會改變匹配到節點的item屬性值，注意node1節點item屬性的變化，如下圖所示。
然後在程式碼#6處執行完後，節點圖示如下，注意紅色箭頭指向的調整。
最後就是消費執行緒喚醒生產執行緒，消費執行緒返回，生產執行緒也返回，過程結束！
好了，原始碼分析就到這裡結束了，你看懂