點擊添加圖片描述（最多60個字）
今天呢！燈塔君跟大家講：
解讀 java 並發隊列 BlockingQueue
最近得空，想寫篇文章好好說說 java 線程池問題，我相信很多人都一知半解的，包括我自己在仔仔細細看源碼之前，也有許多的不解，甚至有些地方我一直都沒有理解到位。
說到線程池實現，那么就不得不涉及到各種 BlockingQueue 的實現，那么我想就 BlockingQueue 的問題和大家分享分享我了解的一些知識。
本文沒有像之前分析 AQS 那樣一行一行源碼分析了，不過還是把其中最重要和最難理解的代碼說了一遍，所以不免篇幅略長。本文涉及到比較多的 Doug Lea 對 BlockingQueue 的設計思想，希望有心的讀者真的可以有一些收獲，我覺得自己還是寫了一些干貨的。
本文直接參考 Doug Lea 寫的 Java doc 和注釋，這也是我們在學習 java 並發包時最好的材料了。希望大家能有所思、有所悟，學習 Doug Lea 的代碼風格，並將其優雅、嚴謹的作風應用到我們寫的每一行代碼中。
目錄：
BlockingQueue
開篇先介紹下 BlockingQueue 這個接口的規則，后面再看其實現。
首先，最基本的來說， BlockingQueue 是一個先進先出的隊列（Queue），為什么說是阻塞（Blocking）的呢？是因為 BlockingQueue 支持當獲取隊列元素但是隊列為空時，會阻塞等待隊列中有元素再返回；也支持添加元素時，如果隊列已滿，那么等到隊列可以放入新元素時再放入。
BlockingQueue 是一個接口，繼承自 Queue，所以其實現類也可以作為 Queue 的實現來使用，而 Queue 又繼承自 Collection 接口。
BlockingQueue 對插入操作、移除操作、獲取元素操作提供了四種不同的方法用於不同的場景中使用：1、拋出異常；2、返回特殊值（null 或 true/false，取決於具體的操作）；3、阻塞等待此操作，直到這個操作成功；4、阻塞等待此操作，直到成功或者超時指定時間。總結如下：
點擊添加圖片描述（最多60個字）
BlockingQueue 的各個實現都遵循了這些規則，當然我們也不用死記這個表格，知道有這么回事，然后寫代碼的時候根據自己的需要去看方法的注釋來選取合適的方法即可。
對於 BlockingQueue，我們的關注點應該在 put(e) 和 take() 這兩個方法，因為這兩個方法是帶阻塞的。
BlockingQueue 不接受 null 值的插入，相應的方法在碰到 null 的插入時會拋出 NullPointerException 異常。null 值在這里通常用於作為特殊值返回（表格中的第三列），代表 poll 失敗。所以，如果允許插入 null 值的話，那獲取的時候，就不能很好地用 null 來判斷到底是代表失敗，還是獲取的值就是 null 值。
一個 BlockingQueue 可能是有界的，如果在插入的時候，發現隊列滿了，那么 put 操作將會阻塞。通常，在這里我們說的無界隊列也不是說真正的無界，而是它的容量是 Integer.MAX_VALUE（21億多）。
BlockingQueue 是設計用來實現生產者-消費者隊列的，當然，你也可以將它當做普通的 Collection 來用，前面說了，它實現了 java.util.Collection 接口。例如，我們可以用 remove(x) 來刪除任意一個元素，但是，這類操作通常並不高效，所以盡量只在少數的場合使用，比如一條消息已經入隊，但是需要做取消操作的時候。
BlockingQueue 的實現都是線程安全的，但是批量的集合操作如 addAll, containsAll, retainAll 和 removeAll 不一定是原子操作。如 addAll(c) 有可能在添加了一些元素后中途拋出異常，此時 BlockingQueue 中已經添加了部分元素，這個是允許的，取決於具體的實現。
BlockingQueue 不支持 close 或 shutdown 等關閉操作，因為開發者可能希望不會有新的元素添加進去，此特性取決於具體的實現，不做強制約束。
最后，BlockingQueue 在生產者-消費者的場景中，是支持多消費者和多生產者的，說的其實就是線程安全問題。
相信上面說的每一句都很清楚了，BlockingQueue 是一個比較簡單的線程安全容器，下面我會分析其具體的在 JDK 中的實現，這里又到了 Doug Lea 表演時間了。
BlockingQueue 實現之 ArrayBlockingQueue
ArrayBlockingQueue 是 BlockingQueue 接口的有界隊列實現類，底層采用數組來實現。
其並發控制采用可重入鎖來控制，不管是插入操作還是讀取操作，都需要獲取到鎖才能進行操作。
如果讀者看過我之前寫的《一行一行源碼分析清楚 AbstractQueuedSynchronizer（二）》 的關於 Condition 的文章的話，那么你一定能很容易看懂 ArrayBlockingQueue 的源碼，它采用一個 ReentrantLock 和相應的兩個 Condition 來實現。
ArrayBlockingQueue 共有以下幾個屬性：
點擊添加圖片描述（最多60個字）
我們用個示意圖來描述其同步機制：
點擊添加圖片描述（最多60個字）
ArrayBlockingQueue 實現並發同步的原理就是，讀操作和寫操作都需要獲取到 AQS 獨占鎖才能進行操作。如果隊列為空，這個時候讀操作的線程進入到讀線程隊列排隊，等待寫線程寫入新的元素，然后喚醒讀線程隊列的第一個等待線程。如果隊列已滿，這個時候寫操作的線程進入到寫線程隊列排隊，等待讀線程將隊列元素移除騰出空間，然后喚醒寫線程隊列的第一個等待線程。
對於 ArrayBlockingQueue，我們可以在構造的時候指定以下三個參數：
- 隊列容量，其限制了隊列中最多允許的元素個數；
- 指定獨占鎖是公平鎖還是非公平鎖。非公平鎖的吞吐量比較高，公平鎖可以保證每次都是等待最久的線程獲取到鎖；
- 可以指定用一個集合來初始化，將此集合中的元素在構造方法期間就先添加到隊列中。
更具體的源碼我就不進行分析了，因為它就是 AbstractQueuedSynchronizer 中 Condition 的使用，感興趣的讀者請看我寫的《一行一行源碼分析清楚 AbstractQueuedSynchronizer（二）》，因為只要看懂了那篇文章，ArrayBlockingQueue 的代碼就沒有分析的必要了，當然，如果你完全不懂 Condition，那么基本上也就可以說看不懂 ArrayBlockingQueue 的源碼了。
BlockingQueue 實現之 LinkedBlockingQueue
底層基於單向鏈表實現的阻塞隊列，可以當做無界隊列也可以當做有界隊列來使用。看構造方法：
點擊添加圖片描述（最多60個字）
點擊添加圖片描述（最多60個字）
我們看看這個類有哪些屬性：
點擊添加圖片描述（最多60個字）
這里用了兩個鎖，兩個 Condition，簡單介紹如下：
takeLock 和 notEmpty 怎么搭配：如果要獲取（take）一個元素，需要獲取 takeLock 鎖，但是獲取了鎖還不夠，如果隊列此時為空，還需要隊列不為空（notEmpty）這個條件（Condition）。
putLock 需要和 notFull 搭配：如果要插入（put）一個元素，需要獲取 putLock 鎖，但是獲取了鎖還不夠，如果隊列此時已滿，還需要隊列不是滿的（notFull）這個條件（Condition）。
首先，這里用一個示意圖來看看 LinkedBlockingQueue 的並發讀寫控制，然后再開始分析源碼：
點擊添加圖片描述（最多60個字）
看懂這個示意圖，源碼也就簡單了，讀操作是排好隊的，寫操作也是排好隊的，唯一的並發問題在於一個寫操作和一個讀操作同時進行，只要控制好這個就可以了。
先上構造方法：
點擊添加圖片描述（最多60個字）
注意，這里會初始化一個空的頭結點，那么第一個元素入隊的時候，隊列中就會有兩個元素。讀取元素時，也總是獲取頭節點后面的一個節點。count 的計數值不包括這個頭節點。
我們來看下 put 方法是怎么將元素插入到隊尾的：
點擊添加圖片描述（最多60個字）
我們再看看 take 方法：
點擊添加圖片描述（最多60個字）
源碼分析就到這里結束了吧，畢竟還是比較簡單的源碼，基本上只要讀者認真點都看得懂。
BlockingQueue 實現之 SynchronousQueue
它是一個特殊的隊列，它的名字其實就蘊含了它的特征 - - 同步的隊列。為什么說是同步的呢？這里說的並不是多線程的並發問題，而是因為當一個線程往隊列中寫入一個元素時，寫入操作不會立即返回，需要等待另一個線程來將這個元素拿走；同理，當一個讀線程做讀操作的時候，同樣需要一個相匹配的寫線程的寫操作。這里的 Synchronous 指的就是讀線程和寫線程需要同步，一個讀線程匹配一個寫線程。
我們比較少使用到 SynchronousQueue 這個類，不過它在線程池的實現類 ThreadPoolExecutor 中得到了應用，感興趣的讀者可以在看完這個后去看看相應的使用。
雖然上面我說了隊列，但是 SynchronousQueue 的隊列其實是虛的，其不提供任何空間（一個都沒有）來存儲元素。數據必須從某個寫線程交給某個讀線程，而不是寫到某個隊列中等待被消費。
你不能在 SynchronousQueue 中使用 peek 方法（在這里這個方法直接返回 null），peek 方法的語義是只讀取不移除，顯然，這個方法的語義是不符合 SynchronousQueue 的特征的。SynchronousQueue 也不能被迭代，因為根本就沒有元素可以拿來迭代的。雖然 SynchronousQueue 間接地實現了 Collection 接口，但是如果你將其當做 Collection 來用的話，那么集合是空的。當然，這個類也是不允許傳遞 null 值的（並發包中的容器類好像都不支持插入 null 值，因為 null 值往往用作其他用途，比如用於方法的返回值代表操作失敗）。
接下來，我們來看看具體的源碼實現吧，它的源碼不是很簡單的那種，我們需要先搞清楚它的設計思想。
源碼加注釋大概有 1200 行，我們先看大框架：
點擊添加圖片描述（最多60個字）
Transferer 有兩個內部實現類，是因為構造 SynchronousQueue 的時候，我們可以指定公平策略。公平模式意味着，所有的讀寫線程都遵守先來后到，FIFO 嘛，對應 TransferQueue。而非公平模式則對應 TransferStack。
點擊添加圖片描述（最多60個字）
我們先采用公平模式分析源碼，然后再說說公平模式和非公平模式的區別。
接下來，我們看看 put 方法和 take 方法：
點擊添加圖片描述（最多60個字）
我們看到，寫操作 put(E o) 和讀操作 take() 都是調用 Transferer.transfer(…) 方法，區別在於第一個參數是否為 null 值。
我們來看看 transfer 的設計思路，其基本算法如下：
- 當調用這個方法時，如果隊列是空的，或者隊列中的節點和當前的線程操作類型一致（如當前操作是 put 操作，而隊列中的元素也都是寫線程）。這種情況下，將當前線程加入到等待隊列即可。
- 如果隊列中有等待節點，而且與當前操作可以匹配（如隊列中都是讀操作線程，當前線程是寫操作線程，反之亦然）。這種情況下，匹配等待隊列的隊頭，出隊，返回相應數據。
其實這里有個隱含的條件被滿足了，隊列如果不為空，肯定都是同種類型的節點，要么都是讀操作，要么都是寫操作。這個就要看到底是讀線程積壓了，還是寫線程積壓了。
我們可以假設出一個男女配對的場景：一個男的過來，如果一個人都沒有，那么他需要等待；如果發現有一堆男的在等待，那么他需要排到隊列后面；如果發現是一堆女的在排隊，那么他直接牽走隊頭的那個女的。
既然這里說到了等待隊列，我們先看看其實現，也就是 QNode:
點擊添加圖片描述（最多60個字）
相信說了這么多以后，我們再來看 transfer 方法的代碼就輕松多了。
點擊添加圖片描述（最多60個字）
點擊添加圖片描述（最多60個字）
Doug Lea 的巧妙之處在於，將各個代碼湊在了一起，使得代碼非常簡潔，當然也同時增加了我們的閱讀負擔，看代碼的時候，還是得仔細想想各種可能的情況。
下面，再說說前面說的公平模式和非公平模式的區別。
相信大家心里面已經有了公平模式的工作流程的概念了，我就簡單說說 TransferStack 的算法，就不分析源碼了。
- 當調用這個方法時，如果隊列是空的，或者隊列中的節點和當前的線程操作類型一致（如當前操作是 put 操作，而棧中的元素也都是寫線程）。這種情況下，將當前線程加入到等待棧中，等待配對。然后返回相應的元素，或者如果被取消了的話，返回 null。
- 如果棧中有等待節點，而且與當前操作可以匹配（如棧里面都是讀操作線程，當前線程是寫操作線程，反之亦然）。將當前節點壓入棧頂，和棧中的節點進行匹配，然后將這兩個節點出棧。配對和出棧的動作其實也不是必須的，因為下面的一條會執行同樣的事情。
- 如果棧頂是進行匹配而入棧的節點，幫助其進行匹配並出棧，然后再繼續操作。
應該說，TransferStack 的源碼要比 TransferQueue 的復雜一些，如果讀者感興趣，請自行進行源碼閱讀。
BlockingQueue 實現之 PriorityBlockingQueue
帶排序的 BlockingQueue 實現，其並發控制采用的是 ReentrantLock，隊列為無界隊列（ArrayBlockingQueue 是有界隊列，LinkedBlockingQueue 也可以通過在構造函數中傳入 capacity 指定隊列最大的容量，但是 PriorityBlockingQueue 只能指定初始的隊列大小，后面插入元素的時候，如果空間不夠的話會自動擴容）。
簡單地說，它就是 PriorityQueue 的線程安全版本。不可以插入 null 值，同時，插入隊列的對象必須是可比較大小的（comparable），否則報 ClassCastException 異常。它的插入操作 put 方法不會 block，因為它是無界隊列（take 方法在隊列為空的時候會阻塞）。
它的源碼相對比較簡單，本節將介紹其核心源碼部分。
我們來看看它有哪些屬性：
點擊添加圖片描述（最多60個字）
此類實現了 Collection 和 Iterator 接口中的所有接口方法，對其對象進行迭代並遍歷時，不能保證有序性。如果你想要實現有序遍歷，建議采用 Arrays.sort(queue.toArray()) 進行處理。PriorityBlockingQueue 提供了 drainTo 方法用於將部分或全部元素有序地填充（准確說是轉移，會刪除原隊列中的元素）到另一個集合中。還有一個需要說明的是，如果兩個對象的優先級相同（compare 方法返回 0），此隊列並不保證它們之間的順序。
PriorityBlockingQueue 使用了基於數組的二叉堆來存放元素，所有的 public 方法采用同一個 lock 進行並發控制。
二叉堆：一顆完全二叉樹，它非常適合用數組進行存儲，對於數組中的元素 a[i]，其左子節點為 a[2*i+1]，其右子節點為 a[2*i + 2]，其父節點為 a[(i-1)/2]，其堆序性質為，每個節點的值都小於其左右子節點的值。二叉堆中最小的值就是根節點，但是刪除根節點是比較麻煩的，因為需要調整樹。
簡單用個圖解釋一下二叉堆，我就不說太多專業的嚴謹的術語了，這種數據結構的優點是一目了然的，最小的元素一定是根元素，它是一棵滿的樹，除了最后一層，最后一層的節點從左到右緊密排列。
點擊添加圖片描述（最多60個字）
下面開始 PriorityBlockingQueue 的源碼分析，首先我們來看看構造方法:
點擊添加圖片描述（最多60個字）
接下來，我們來看看其內部的自動擴容實現：
點擊添加圖片描述（最多60個字）
擴容方法對並發的控制也非常的巧妙，釋放了原來的獨占鎖 lock，這樣的話，擴容操作和讀操作可以同時進行，提高吞吐量。
下面，我們來分析下寫操作 put 方法和讀操作 take 方法。
點擊添加圖片描述（最多60個字）
對於二叉堆而言，插入一個節點是簡單的，插入的節點如果比父節點小，交換它們，然后繼續和父節點比較。
點擊添加圖片描述（最多60個字）
我們用圖來示意一下，我們接下來要將 11 插入到隊列中，看看 siftUp 是怎么操作的。
點擊添加圖片描述（最多60個字）
我們再看看 take 方法：
點擊添加圖片描述（最多60個字）
dequeue 方法返回隊頭，並調整二叉堆的樹，調用這個方法必須先獲取獨占鎖。
廢話不多說，出隊是非常簡單的，因為隊頭就是最小的元素，對應的是數組的第一個元素。難點是隊頭出隊后，需要調整樹。
點擊添加圖片描述（最多60個字）
點擊添加圖片描述（最多60個字）
記住二叉堆是一棵完全二叉樹，那么根節點 10 拿掉后，最后面的元素 17 必須找到合適的地方放置。首先，17 和 10 不能直接交換，那么先將根節點 10 的左右子節點中較小的節點往上滑，即 12 往上滑，然后原來 12 留下了一個空節點，然后再把這個空節點的較小的子節點往上滑，即 13 往上滑，最后，留出了位子，17 補上即可。
我稍微調整下這個樹，以便讀者能更明白：
點擊添加圖片描述（最多60個字）
好了， PriorityBlockingQueue 我們也說完了。
總結
我知道本文過長，相信一字不漏看完的讀者肯定是少數。
ArrayBlockingQueue 底層是數組，有界隊列，如果我們要使用生產者-消費者模式，這是非常好的選擇。
LinkedBlockingQueue 底層是鏈表，可以當做無界和有界隊列來使用，所以大家不要以為它就是無界隊列。
SynchronousQueue 本身不帶有空間來存儲任何元素，使用上可以選擇公平模式和非公平模式。
PriorityBlockingQueue 是無界隊列，基於數組，數據結構為二叉堆，數組第一個也是樹的根節點總是最小值。