Java併發之BlockingQueue

時間 2019-12-08

原文原文鏈接

1、Queue編程

Queue是隊列接口是 Collection的子接口。除了基本的 Collection操做外，隊列還提供其餘的插入、提取和檢查操做。每一個方法都存在兩種形式：一種拋出異常（操做失敗時），另外一種返回一個特殊值（null 或 false，具體取決於操做）。插入操做的後一種形式是用於專門爲有容量限制的 Queue 實現設計的；在大多數實現中，插入操做不會失敗。緩存

	拋出異常	返回特殊值
插入	add(e)	offer(e)
移除	remove(e)	poll(e)
檢查	element()	peek()

隊列一般（但並不是必定）以 FIFO（先進先出）的方式排序各個元素。不過優先級隊列和 LIFO 隊列（或堆棧）例外，前者根據提供的比較器或元素的天然順序對元素進行排序，後者按 LIFO（後進先出）的方式對元素進行排序。不管使用哪一種排序方式，隊列的頭都是調用 remove() 或 poll() 所移除的元素。在 FIFO 隊列中，全部的新元素都插入隊列的末尾。其餘種類的隊列可能使用不一樣的元素放置規則。每一個 Queue 實現必須指定其順序屬性。安全

若是可能，offer 方法可插入一個元素，失敗則返回 false。這與 Collection.add 方法不一樣，該方法只能經過拋出未經檢查的異常使添加元素失敗。offer 方法設計用於正常的失敗狀況，而不是出現異常的狀況，例如在容量固定（有界）的隊列中。併發

remove() 和 poll() 方法可移除和返回隊列的頭。到底從隊列中移除哪一個元素是隊列排序策略的功能，而該策略在各類實現中是不一樣的。remove() 和 poll() 方法僅在隊列爲空時其行爲有所不一樣：remove() 方法拋出一個異常，而 poll() 方法則返回 null。學習

element() 和 peek() 獲取但不移除隊列的頭，element與 peek 惟一的不一樣在於：此隊列爲空時將拋出一個異常。this

Queue 接口並未定義阻塞隊列的方法，而這在併發編程中是很常見的。BlockingQueue 接口則定義了那些等待元素出現或等待隊列中有可用空間的方法，這些方法擴展了此接口。 spa

Queue 實現一般不容許插入 null 元素，儘管某些實現（如 LinkedList）並不由止插入 null。即便在容許 null 的實現中，也不該該將 null 插入到 Queue 中，由於 null 也用做 poll 方法的一個特殊返回值，代表隊列不包含元素。線程

Queue 實現一般未定義 equals 和 hashCode 方法的基於元素的版本，而是從 Object 類繼承了基於身份的版本，由於對於具備相同元素但有不一樣排序屬性的隊列而言，基於元素的相等性並不是老是定義良好的。設計

Queue 做爲隊列能夠實現一個按固定順序訪問其內部元素的結構，與 LinkedList等實現不一樣，Queue並不能獲取指定位置的元素。code

在 ThreadPoolExecutor類中建立線程池時使用的是 BlockingQueue。BlockingQueue是 Queue的子接口，BlockingQueue的實現類有不少：ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue。

Deque與 Queue不一樣在於，Deque是一個雙端隊列，支持在兩端插入和移除元素。名稱 deque 是「double ended queue（雙端隊列）」的縮寫，一般讀爲「deck」。大多數 Deque 實現對於它們可以包含的元素數沒有固定限制，但此接口既支持有容量限制的雙端隊列，也支持沒有固定大小限制的雙端隊列。 Deque不是咱們要學習的重點，下面就不提了。

咱們要用到的實現爲 ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue,DelayedWorkQueue.其中 DelayedWorkQueue是 ScheduledThreadPoolExecutor的內部類實現。

頂層接口爲 Queue，而後是 Queue的抽象實現類 AbstractQueue和子接口 BlockingQueue。圖中四個類均繼承於 AbstractQueue並實現 BlockingQueue接口，DelayedWorkQueue一樣實現了 BlockingQueue，但DelayedWorkQueue繼承自 AbstractCollection。

如下是Queue的源代碼：

Java代碼

public interface Queue<E> extends Collection<E> {
/**
* 將指定的元素插入此隊列（若是當即可行且不會違反容量限制），在成功時返回 true，若是當前沒有可用的空間，則拋出IllegalStateException
*/
boolean add(E e);
/**
* 將指定的元素插入此隊列（若是當即可行且不會違反容量限制），當使用有容量限制的隊列時，此方法一般要優於add(E)，後者可能沒法插入元素，而只是拋出一個異常
*/
boolean offer(E e);
/**
* 獲取並移除此隊列的頭。此方法與 poll 惟一的不一樣在於：此隊列爲空時將拋出一個異常
*/
E remove();
/**
* 獲取並移除此隊列的頭，若是此隊列爲空，則返回 null
*/
E poll();
/**
* 獲取，可是不移除此隊列的頭。此方法與 peek 惟一的不一樣在於：此隊列爲空時將拋出一個異常
*/
E element();
/**
* 獲取但不移除此隊列的頭；若是此隊列爲空，則返回 null
*/
E peek();
}

2、AbstractQueue

AbstractQueue提供某些 Queue 操做的主要實現。此類中的實現適用於基本實現不容許包含 null 元素時。add、remove 和 element 方法分別基於 offer、poll 和 peek 方法，可是它們經過拋出異常而不是返回 false 或 null 來指示失敗。

擴展此類的 Queue 實現至少必須定義一個不容許插入 null 元素的 Queue.offer(E) 方法，該方法以及 Queue.peek()、Queue.poll()、Collection.size() 和 Collection.iterator() 都支持 Iterator.remove() 方法。一般還要重寫其餘方法。若是沒法知足這些要求，那麼能夠轉而考慮爲 AbstractCollection 建立子類。

如下是 AbstractQueue的源代碼：

Java代碼

public abstract class AbstractQueue<E> extends AbstractCollection<E> implements Queue<E> {
/**
* 子類使用的構造方法
*/
protected AbstractQueue() {
}
/**
* 將指定的元素插入到此隊列中（若是當即可行且不會違反容量限制），在成功時返回 true，若是當前沒有可用空間，則拋出 IllegalStateException。
*/
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
/**
* 獲取並移除此隊列的頭。此方法與 poll 惟一的不一樣在於：此隊列爲空時將拋出一個異常。
* 除非隊列爲空，不然此實現返回 poll 的結果。
*/
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
/**
* 獲取但不移除此隊列的頭。此方法與 peek 惟一的不一樣在於：此隊列爲空時將拋出一個異常。
* 除非隊列爲空，不然此實現返回 peek 的結果。
*/
public E element() {
E x = peek();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
/**
* 移除此隊列中的全部元素。此調用返回後，隊列將爲空。
* 此實現重複調用 poll，直到它返回 null 爲止。
*/
public void clear() {
while (poll() != null)
;
}
/**
* 將指定 collection 中的全部元素都添加到此隊列中。
* 若是試圖將某一隊列 addAll 到該隊列自己中，則會致使 IllegalArgumentException。
* 此外，若是正在進行此操做時修改指定的 collection，則此操做的行爲是不肯定的。
* 此實如今指定的 collection 上進行迭代，並依次將迭代器返回的每個元素添加到此隊列中。
* 在試圖添加某一元素（尤爲是 null 元素）時若是遇到了運行時異常，則可能致使在拋出相關異常時只成功地添加了某些元素。
*/
public boolean addAll(Collection<? extends E> c) {
if (c == null)
throw new NullPointerException();
if (c == this)
throw new IllegalArgumentException();
boolean modified = false;
Iterator<? extends E> e = c.iterator();
while (e.hasNext()) {
if (add(e.next()))
modified = true;
}
return modified;
}
}

3、BlockingQueue

阻塞隊列（BlockingQueue）是一個支持兩個附加操做的隊列。這兩個附加的操做是：在隊列爲空時，獲取元素的線程會等待隊列變爲非空。當隊列滿時，存儲元素的線程會等待隊列可用，從而產生阻塞。阻塞隊列經常使用於生產者和消費者的場景，生產者是往隊列裏添加元素的線程，消費者是從隊列裏拿元素的線程。阻塞隊列就是生產者存放元素的緩存容器，而消費者也只從容器裏拿元素。

BlockingQueue 的方法以四種形式出現，這四種形式的處理方式不一樣：第一種是拋出一個異常，第二種是返回一個特殊值（null 或 false，具體取決於操做），第三種是在操做能夠成功前，無限期地阻塞當前線程，第四種是在放棄前只在給定的最大時間限制內阻塞。下表中總結了這些方法：

	拋出異常	返回特殊值	阻塞	超時
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除	remove()	poll()	take()	poll(time, unit)
檢查	element()	peek()	-	-

• 拋出異常：是指當阻塞隊列滿時候，再往隊列裏插入元素，會拋出IllegalStateException("Queue full")異常。當隊列爲空時，從隊列裏獲取元素時會拋出NoSuchElementException異常。

• 返回特殊值：插入方法會返回是否成功，成功則返回true。移除方法，則是從隊列裏拿出一個元素，若是沒有則返回null

• 阻塞：當阻塞隊列滿時，若是生產者線程往隊列裏put元素，隊列會一直阻塞生產者線程，直到拿到數據，或者響應中斷退出。當隊列空時，消費者線程試圖從隊列裏take元素，隊列也會阻塞消費者線程，直到隊列可用。

• 超時：當阻塞隊列滿時，隊列會阻塞生產者線程一段時間，若是超過必定的時間，生產者線程就會退出。

BlockingQueue 不接受 null 元素。試圖 add、put 或 offer 一個 null 元素時，某些實現會拋出 NullPointerException。null 被用做指示 poll 操做失敗的警惕值。

BlockingQueue 能夠是限定容量的。它在任意給定時間均可以有一個 remainingCapacity，超出此容量，便沒法無阻塞地 put 附加元素。沒有任何內部容量約束的 BlockingQueue 老是報告 Integer.MAX_VALUE 的剩餘容量。

BlockingQueue 實現主要用於生產者-消費者隊列，但它另外還支持 Collection 接口。所以，舉例來講，使用 remove(x) 從隊列中移除任意一個元素是有可能的。然而，這種操做一般不會有效執行，只能有計劃地偶爾使用，好比在取消排隊信息時。

題外話：所謂生產者消費者模式，這裏簡單介紹一下。好比咱們在餐廳吃飯，咱們就是消費者，餐廳的廚師就是生產者，而餐廳的服務員就是一個緩衝環節。當生產者製做好菜品（生產產品），交由服務員（緩衝區），由服務員將菜品送至顧客（消費者）品用。

　　生產者-消費者模式最重要的做用就是解耦，利用緩衝區將二者分離。若是每個廚師作完了菜都須要親自送到顧客桌上，那麼這樣就是將廚師與顧客綁定到了一塊兒。加入中間緩衝環節，也就是服務員，將送菜的任務交由服務員（緩衝區）去處理，這樣生產者與消費者就能夠各自作本身的事情了。在此模式下二者間支持併發操做，由於飯店的廚師確定不止一個，顧客也是如此。再有就是支持二者間不一樣步，由於二者間的數量與效率是不一樣步的，這就會致使生產與消費的速度不一樣。

BlockingQueue 實現是線程安全的。全部排隊方法均可以使用內部鎖或其餘形式的併發控制來自動達到它們的目的。然而，大量的 Collection 操做（addAll、containsAll、retainAll 和 removeAll）沒有必要自動執行，除非在實現中特別說明。所以，舉例來講，在只添加了 c 中的一些元素後，addAll(c) 有可能失敗（拋出一個異常）。

BlockingQueue 實質上不支持使用任何一種「close」或「shutdown」操做來指示再也不添加任何項。這種功能的需求和使用有依賴於實現的傾向。例如，一種經常使用的策略是：對於生產者，插入特殊的 end-of-stream 或 poison 對象，並根據使用者獲取這些對象的時間來對它們進行解釋。

注意，BlockingQueue 能夠安全地與多個生產者和多個使用者一塊兒使用。

如下是基於典型的生產者-消費者場景的一個用例：

Java代碼