解讀 java 併發隊列 BlockingQueue

 

點擊添加圖片描述（最多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 對插入操做、移除操做、獲取元素操做提供了四種不一樣的方法用於不一樣的場景中使用：一、拋出異常；二、返回特殊值（null 或 true/false，取決於具體的操做）；三、阻塞等待此操做，直到這個操做成功；四、阻塞等待此操做，直到成功或者超時指定時間。總結以下：

 

點擊添加圖片描述（最多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，咱們能夠在構造的時候指定如下三個參數：

1. 隊列容量，其限制了隊列中最多容許的元素個數；
2. 指定獨佔鎖是公平鎖仍是非公平鎖。非公平鎖的吞吐量比較高，公平鎖能夠保證每次都是等待最久的線程獲取到鎖；
3. 能夠指定用一個集合來初始化，將此集合中的元素在構造方法期間就先添加到隊列中。

更具體的源碼我就不進行分析了，由於它就是 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 的設計思路，其基本算法以下：

1. 當調用這個方法時，若是隊列是空的，或者隊列中的節點和當前的線程操做類型一致（如當前操做是 put 操做，而隊列中的元素也都是寫線程）。這種狀況下，將當前線程加入到等待隊列便可。
2. 若是隊列中有等待節點，並且與當前操做能夠匹配（如隊列中都是讀操做線程，當前線程是寫操做線程，反之亦然）。這種狀況下，匹配等待隊列的隊頭，出隊，返回相應數據。

其實這裏有個隱含的條件被知足了，隊列若是不爲空，確定都是同種類型的節點，要麼都是讀操做，要麼都是寫操做。這個就要看究竟是讀線程積壓了，仍是寫線程積壓了。

咱們能夠假設出一個男女配對的場景：一個男的過來，若是一我的都沒有，那麼他須要等待；若是發現有一堆男的在等待，那麼他須要排到隊列後面；若是發現是一堆女的在排隊，那麼他直接牽走隊頭的那個女的。

既然這裏說到了等待隊列，咱們先看看其實現，也就是 QNode:

點擊添加圖片描述（最多60個字）編輯

相信說了這麼多之後，咱們再來看 transfer 方法的代碼就輕鬆多了。

點擊添加圖片描述（最多60個字）編輯

 

點擊添加圖片描述（最多60個字）編輯

 

Doug Lea 的巧妙之處在於，將各個代碼湊在了一塊兒，使得代碼很是簡潔，固然也同時增長了咱們的閱讀負擔，看代碼的時候，仍是得仔細想一想各類可能的狀況。

下面，再說說前面說的公平模式和非公平模式的區別。

相信你們內心面已經有了公平模式的工做流程的概念了，我就簡單說說 TransferStack 的算法，就不分析源碼了。

1. 當調用這個方法時，若是隊列是空的，或者隊列中的節點和當前的線程操做類型一致（如當前操做是 put 操做，而棧中的元素也都是寫線程）。這種狀況下，將當前線程加入到等待棧中，等待配對。而後返回相應的元素，或者若是被取消了的話，返回 null。
2. 若是棧中有等待節點，並且與當前操做能夠匹配（如棧裏面都是讀操做線程，當前線程是寫操做線程，反之亦然）。將當前節點壓入棧頂，和棧中的節點進行匹配，而後將這兩個節點出棧。配對和出棧的動做其實也不是必須的，由於下面的一條會執行一樣的事情。
3. 若是棧頂是進行匹配而入棧的節點，幫助其進行匹配並出棧，而後再繼續操做。

應該說，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 是無界隊列，基於數組，數據結構爲二叉堆，數組第一個也是樹的根節點老是最小值。