解讀 Java 併發隊列 BlockingQueue

BlockingQueue 是一個先進先出的隊列（Queue）

BlockingQueue 支持當獲取隊列元素可是隊列爲空時，會阻塞等待隊列中有元素再返回；

                    也支持添加元素時，若是隊列已滿，那麼等到隊列能夠放入新元素時再放入。

 

BlockingQueue 對插入操做、移除操做、獲取元素操做提供了四種不一樣的方法用於不一樣的場景中使用：

    一、拋出異常；

    二、返回特殊值（null 或 true/false，取決於具體的操做）；

    三、阻塞等待此操做，直到這個操做成功；

    四、阻塞等待此操做，直到成功或者超時指定時間。

 

BlockingQueue，咱們的關注點應該在 put(e) 和 take() 這兩個方法，由於這兩個方法是帶阻塞的。

一個 BlockingQueue 多是有界的，若是在插入的時候，發現隊列滿了，那麼 put 操做將會阻塞。

一般，在這裏咱們說的無界隊列也不是說真正的無界，而是它的容量是 Integer.MAX_VALUE（21億多）。

BlockingQueue 是設計用來實現生產者-消費者隊列的

 

BlockingQueue 的實現都是線程安全的，可是批量的集合操做如 addAll, containsAll, retainAll 和 removeAll 不必定是原子操做。如 addAll(c) 有可能在添加了一些元素後中途拋出異常，此時 BlockingQueue 中已經添加了部分元素，這個是容許的，取決於具體的實現。

 

BlockingQueue 不支持 close 或 shutdown 等關閉操做，由於開發者可能但願不會有新的元素添加進去，此特性取決於具體的實現，不作強制約束。

 

ArrayBlockingQueue 是 BlockingQueue 接口的有界隊列實現類，底層採用數組來實現。

其併發控制採用可重入鎖來控制，不論是插入操做仍是讀取操做，都須要獲取到鎖才能進行操做。

 

LinkedBlockingQueue底層基於單向鏈表實現的阻塞隊列，能夠當作無界隊列也能夠當作有界隊列來使用。

它的實現用了兩個鎖，兩個 Condition，簡單介紹以下：

takeLock 和 notEmpty 怎麼搭配：若是要獲取（take）一個元素，須要獲取 takeLock 鎖，可是獲取了鎖還不夠，若是隊列此時爲空，還須要隊列不爲空（notEmpty）這個條件（Condition）。

putLock 須要和 notFull 搭配：若是要插入（put）一個元素，須要獲取 putLock 鎖，可是獲取了鎖還不夠，若是隊列此時已滿，還須要隊列不是滿的（notFull）這個條件（Condition）。

 

SynchronousQueue同步的隊列。由於當一個線程往隊列中寫入一個元素時，寫入操做不會當即返回，須要等待另外一個線程來將這個元素拿走；同理，當一個讀線程作讀操做的時候，一樣須要一個相匹配的寫線程的寫操做。這裏的 Synchronous 指的就是讀線程和寫線程須要同步，一個讀線程匹配一個寫線程。

SynchronousQueue 的隊列實際上是虛的，其不提供任何空間（一個都沒有）來存儲元素。數據必須從某個寫線程交給某個讀線程，而不是寫到某個隊列中等待被消費。

Transferer 有兩個內部實現類，是由於構造 SynchronousQueue 的時候，咱們能夠指定公平策略。公平模式意味着，全部的讀寫線程都遵照先來後到，FIFO 嘛，對應 TransferQueue。而非公平模式則對應 TransferStack。

 

TransferQueue：（interface TransferQueue<E> extends BlockingQueue<E>） 也是一個阻塞隊列，它具有阻塞隊列的全部特性。LinkedTransferQueue

    1.transfer(E e)若當前存在一個正在等待獲取的消費者線程，即馬上將e移交之；不然將元素e插入到隊列尾部，而且當前線程進入阻塞狀態，直到有消費者線程取走該元素。

    2.tryTransfer(E e)若當前存在一個正在等待獲取的消費者線程，則該方法會即刻轉移e，並返回true;若不存在則返回false，可是並不會將e插入到隊列中。這個方法不會阻塞當前線程，要麼快速返回true，要麼快速返回false。    

    3.hasWaitingConsumer()和getWaitingConsumerCount()用來判斷當前正在等待消費的消費者線程個數。

    4.tryTransfer(E e, long timeout, TimeUnit unit) 若當前存在一個正在等待獲取的消費者線程，會當即傳輸給它; 不然將元素e插入到隊列尾部，而且等待被消費者線程獲取消費掉。若在指定的時間內元素e沒法被消費者線程獲取，則返回false，同時該元素從隊列中移除。

https://blog.csdn.net/aitangyong/article/details/46472643

http://ifeve.com/java-transfer-queue/

 

PriorityBlockingQueue帶排序的 BlockingQueue 實現，其併發控制採用的是 ReentrantLock，隊列爲無界隊列（ArrayBlockingQueue 是有界隊列，LinkedBlockingQueue 也能夠經過在構造函數中傳入 capacity 指定隊列最大的容量，可是 PriorityBlockingQueue 只能指定初始的隊列大小，後面插入元素的時候，若是空間不夠的話會自動擴容）。

簡單地說，它就是 PriorityQueue 的線程安全版本。

 

總結：

ArrayBlockingQueue 底層是數組，有界隊列，若是咱們要使用生產者-消費者模式，這是很是好的選擇。

LinkedBlockingQueue 底層是鏈表，能夠當作無界和有界隊列來使用，因此你們不要覺得它就是無界隊列。

SynchronousQueue 自己不帶有空間來存儲任何元素，使用上能夠選擇公平模式和非公平模式。

PriorityBlockingQueue 是無界隊列，基於數組，數據結構爲二叉堆，數組第一個也是樹的根節點老是最小值。

 

https://mp.weixin.qq.com/s/UlgoDIvR1FdpiKfbeeW0aw

 

LinkedBlockingQueue 與 ConcurrentLinkedQueue區別：

Concurrent*的容器真正表明併發， *BlockingQueue 表明阻塞。

Concurrent類型基於lock-free，在常見的多線程訪問場景，能夠提供較高吞吐量，但size是弱一致性，不夠精準。

Concurrent類的沒有CopyOnWrite類的容器有交重的修改開銷。

BlockingQueue 內部基於鎖，經過阻塞特性。