系列傳送門：
- Java並發包源碼學習系列：AbstractQueuedSynchronizer
- Java並發包源碼學習系列：CLH同步隊列及同步資源獲取與釋放
- Java並發包源碼學習系列：AQS共享式與獨占式獲取與釋放資源的區別
- Java並發包源碼學習系列：ReentrantLock可重入獨占鎖詳解
- Java並發包源碼學習系列：ReentrantReadWriteLock讀寫鎖解析
- Java並發包源碼學習系列：詳解Condition條件隊列、signal和await
- Java並發包源碼學習系列：掛起與喚醒線程LockSupport工具類
- Java並發包源碼學習系列：JDK1.8的ConcurrentHashMap源碼解析
本篇要點
- 介紹阻塞隊列的概述：支持阻塞式插入和移除的隊列結構。
- 介紹阻塞隊列提供的方法。
- 介紹BlockingQueue接口的幾大實現類及主要特點。
- 以ArrayBlockingQueue為例介紹等待通知實現阻塞隊列的過程。
不會涉及到太多源碼部分，意在對阻塞隊列章節的全局概覽進行總結，具體的每種具體實現，之后會一一分析學習。
什么是阻塞隊列
阻塞隊列 = 阻塞 + 隊列。
-
隊列：一種先進先出的數據結構，支持尾部添加、首部移除或查看等基礎操作。
-
阻塞：除了隊列提供的基本操作之外，還提供了支持阻塞式插入和移除的方式。
下面這些對BlockingQueue的介紹基本翻譯自JavaDoc，非常詳細。
- 阻塞隊列的頂級接口是
java.util.concurrent.BlockingQueue,它繼承了Queue，Queue又繼承自Collection接口。 - BlockingQueue 對插入操作、移除操作、獲取元素操作提供了四種不同的方法用於不同的場景中使用：1、拋出異常；2、返回特殊值（null 或 true/false，取決於具體的操作）；3、阻塞等待此操作，直到這個操作成功；4、阻塞等待此操作，直到成功或者超時指定時間，第二節會有詳細介紹。
- BlockingQueue不接受null的插入，否則將拋出空指針異常，因為poll失敗了會返回null，如果允許插入null值，就無法判斷poll是否成功了。
- BlockingQueue可能是有界的，如果在插入的時候發現隊列滿了，將會阻塞，而無界隊列則有
Integer.MAX_VALUE大的容量，並不是真的無界。 - BlockingQueue通常用來作為生產者-消費者的隊列的，但是它也支持Collection接口提供的方法，比如使用remove(x)來刪除一個元素，但是這類操作並不是很高效，因此盡量在少數情況下使用，如：當一條入隊的消息需要被取消的時候。
- BlockingQueue的實現都是線程安全的，所有隊列的操作或使用內置鎖或是其他形式的並發控制來保證原子。但是一些批量操作如：
addAll,containsAll,retainAll和removeAll不一定是原子的。如 addAll(c) 有可能在添加了一些元素后中途拋出異常，此時 BlockingQueue 中已經添加了部分元素。 - BlockingQueue不支持類似close或shutdown等關閉操作。
下面這一段是並發大師 DougLea 寫的一段demo，使用BlockingQueue 來保證多生產者和消費者時的線程安全
// Doug Lea： BlockingQueue 可以用來保證多生產者和消費者時的線程安全
class Producer implements Runnable{
private final BlockingQueue queue;
Producer(BlockingQueue q){
queue = q;
}
public void run(){
try{
while(true) {
queue.put(produce()); // 阻塞式插入
}
}catch(InterruptedException ex){ ...handle... }
}
Object produce() { ... }
}
class Consumer implements Runnable{
private final BlockingQueue queue;
Consumer(BlockingQueue q){
queue = q;
}
public void run(){
try{
while(true) {
consume(queue.take())); // 阻塞式獲取
}
}catch(InterruptedException ex){ ...handle... }
}
void consume(Object x) { ... }
}
class Setup{
void main(){
BlockingQueue q = new SomeQueueImplementation();
Producer p = new Producer(q);
Consumer c1 = new Consumer(q);
Consumer c2 = new Consumer(q);
new Thread(p).start();
new Thread(c1).start();
new Thread(c2).start();
}
}
阻塞隊列提供的方法
BlockingQueue 對插入操作、移除操作、獲取元素操作提供了四種不同的方法用於不同的場景中使用：
| 方法類別 | 拋出異常 | 返回特殊值 | 一直阻塞 | 超時退出 |
|---|---|---|---|---|
| 插入 | add(e) | offer(e) | put(e) |
offer(e, time, unit) |
| 移除 | remove() | poll() | take() |
poll(time, unit) |
| 瞅一瞅 | element() | peek() |
博主在這邊大概解釋一下，如果隊列可用時，上面的幾種方法其實效果都差不多，但是當隊列空或滿時，會表現出部分差異：
- 拋出異常：當隊列滿時，如果再往隊列里add插入元素e時，會拋出
IllegalStateException: Queue full的異常，如果隊空時，往隊列中取出元素【移除或瞅一瞅】會拋出NoSuchElementException異常。 - 返回特殊值：隊列滿時，offer插入失敗返回false。隊列空時，poll取出元素失敗返回null，而不是拋出異常。
- 一直阻塞：當隊列滿時，put試圖插入元素，將會一直阻塞插入的生產者線程，同理，隊列為空時，如果消費者線程從隊列里take獲取元素，也會阻塞，知道隊列不為空。
- 超時退出：可以理解為一直阻塞情況的超時版本，線程阻塞一段時間，會自動退出阻塞。
我們本篇的重點是阻塞隊列，那么【一直阻塞】和【超時退出】相關的方法是我們分析的重頭啦。
阻塞隊列的七種實現
- ArrayBlockingQueue：由數組構成的有界阻塞隊列。
- LinkedBlockingQueue：由鏈表構成的界限可選的阻塞隊列，如不指定邊界，則為
Integer.MAX_VALUE。 - PriorityBlockingQueue：支持優先級排序【類似於PriorityQueue的排序規則】的無界阻塞隊列。
- DelayQueue：支持延遲獲取元素的無界阻塞隊列。
- SynchronousQueue：不存儲元素的阻塞隊列，每個插入的操作必須等待另一個線程進行相應的刪除操作，反之亦然。
另外BlockingQueue有兩個繼承子接口，分別是：TransferQueue和BlockingDeque，他們有各自的實現類：
- LinkedTransferQueue：由鏈表組成的無界TransferQueue。
- LinkedBlockingDeque：由鏈表構成的界限可選的雙端阻塞隊列，如不指定邊界，則為
Integer.MAX_VALUE。
BlockingDeque比較好理解一些，支持雙端操作嘛，TransferQueue又是個啥玩意呢？
TransferQueue和BlockingQueue的區別
BlockingQueue：當生產者向隊列添加元素但隊列已滿時，生產者會被阻塞；當消費者從隊列移除元素但隊列為空時，消費者會被阻塞。
TransferQueue則更進一步，生產者會一直阻塞直到所添加到隊列的元素被某一個消費者所消費（不僅僅是添加到隊列里就完事）。新添加的transfer方法用來實現這種約束。顧名思義，阻塞就是發生在元素從一個線程transfer到另一個線程的過程中，它有效地實現了元素在線程之間的傳遞（以建立Java內存模型中的happens-before關系的方式）。
1、ArrayBlockingQueue
ArrayBlockingQueue是由數組構成的有界阻塞隊列，支持FIFO的次序對元素進行排序。
這是一個典型的有界緩沖結構，可指定大小存儲元素，供生產線程插入，供消費線程獲取，但注意，容量一旦指定，便不可修改。
隊列空時嘗試take操作和隊列滿時嘗試put操作都會阻塞執行操作的線程。
該類還支持可供選擇的公平性策略，ReentrantLock可重入鎖實現，默認采用非公平策略，當隊列可用時，阻塞的線程都可以爭奪訪問隊列的資格。
// 創建采取公平策略且規定容量為10 的ArrayBlockingQueue
ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10, true);
2、LinkedBlockingQueue
LinkedBlockingQueue是由鏈表構成的界限可選的阻塞隊列，如不指定邊界，則為Integer.MAX_VALUE,因此如不指定邊界，一般來說，插入的時候都會成功。
LinkedBlockingQueue支持FIFO先進先出的次序對元素進行排序。
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
3、PriorityBlockingQueue
PriorityBlockingQueue是一個支持優先級的無界阻塞隊列，基於數組的二叉堆，其實就是線程安全的PriorityQueue。
默認情況下元素采取自然順序升序排列，也可以自定義類實現compareTo()方法來指定元素排序規則，或者初始化PriorityBlockingQueue時，指定構造參數Comparator來對元素進行排序。
需要注意的是如果兩個對象的優先級相同（compare 方法返回 0），此隊列並不保證它們之間的順序。
PriorityBlocking可以傳入一個初始容量，其實也就是底層數組的最小容量，之后會使用grow擴容。
// 這里傳入10是初始容量，之后會擴容啊，無界的~ ， 后面參數可以傳入比較規則，可以用lambda表達式哦
PriorityQueue<Integer> priorityQueue = new PriorityQueue<>(10, new Comparator<Integer>() {
@Override
public int compare (Integer o1, Integer o2) {
return 0;
}
});
4、DelayQueue
DelayQueue是一個支持延時獲取元素的無界阻塞隊列，使用PriorityQueue來存儲元素。
隊中的元素必須實現Delayed接口【Delay接口又繼承了Comparable，需要實現compareTo方法】，每個元素都需要指明過期時間，通過getDelay(unit)獲取元素剩余時間【剩余時間 = 到期時間 - 當前時間】。
當從隊列獲取元素時，只有過期的元素才會出隊列。
static class DelayedElement implements Delayed {
private final long delayTime; // 延遲時間
private final long expire; // 到期時間
private final String taskName; // 任務名稱
public DelayedElement (long delayTime, String taskName) {
this.delayTime = delayTime;
this.taskName = taskName;
expire = now() + delayTime;
}
// 獲取當前時間
final long now () {
return System.currentTimeMillis();
}
// 剩余時間 = 到期時間 - 當前時間
@Override
public long getDelay (TimeUnit unit) {
return unit.convert(expire - now(), TimeUnit.MILLISECONDS);
}
// 靠前的元素是最快過期的元素
@Override
public int compareTo (Delayed o) {
return (int) (getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
}
}
5、SynchronousQueue
SynchronousQueue是一個不存儲元素的阻塞隊列，每個插入的操作必須等待另一個線程進行相應的刪除操作，反之亦然，因此這里的Synchronous指的是讀線程和寫線程需要同步，一個讀線程匹配一個寫線程。
你不能在該隊列中使用peek方法，因為peek是只讀取不移除，不符合該隊列特性，該隊列不存儲任何元素，數據必須從某個寫線程交給某個讀線程，而不是在隊列中等待倍消費,非常適合傳遞性場景。
SynchronousQueue的吞吐量高於LinkedBlockingQueue和ArrayBlockingQueue。
該類還支持可供選擇的公平性策略，默認采用非公平策略，當隊列可用時，阻塞的線程都可以爭奪訪問隊列的資格。
public SynchronousQueue() {
this(false);
}
// 公平策略使用TransferQueue 實現， 非公平策略使用TransferStack 實現
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
6、LinkedTransferQueue
LinkedTransferQueue是由鏈表組成的無界TransferQueue，相對於其他阻塞隊列，多了tryTransfer和transfer方法。
TransferQueue：生產者會一直阻塞直到所添加到隊列的元素被某一個消費者所消費（不僅僅是添加到隊列里就完事）。新添加的transfer方法用來實現這種約束。顧名思義，阻塞就是發生在元素從一個線程transfer到另一個線程的過程中，它有效地實現了元素在線程之間的傳遞（以建立Java內存模型中的happens-before關系的方式）。
7、LinkedBlockingDeque
LinkedBlockingDeque是由鏈表構成的界限可選的雙端阻塞隊列，支持從兩端插入和移除元素，如不指定邊界，則為Integer.MAX_VALUE。
阻塞隊列的實現機制
本文不會過於詳盡地解析每個阻塞隊列源碼實現，但會總結通用的阻塞隊列的實現機制。
以阻塞隊列接口BlockingQueue為例，我們以其中新增的阻塞相關的兩個方法為主要解析對象，put和take方法。
- put：如果隊列已滿，生產者線程便一直阻塞，直到隊列不滿。
- take：如果隊列已空，消費者線程便開始阻塞，直到隊列非空。
其實我們之前在學習Condition的時候已經透露過一些內容，這里利用ReentrantLock實現鎖語義，通過鎖關聯的condition條件隊列來靈活地實現等待通知機制。
之前已經詳細地學習過：Java並發包源碼學習系列：詳解Condition條件隊列、signal和await
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
// 初始化ReentrantLock
lock = new ReentrantLock(fair);
// 創建條件對象
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
put方法
public void put(E e) throws InterruptedException {
// 不能加null 啊
checkNotNull(e);
final ReentrantLock lock = this.lock;
// 可響應中斷地獲取鎖
lock.lockInterruptibly();
try {
// 如果隊列滿了 notFull陷入阻塞，直到signal
while (count == items.length)
notFull.await();
// 如果隊列沒滿，執行入隊操作
enqueue(e);
} finally {
// 解鎖
lock.unlock();
}
}
// 入隊操作
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
// 注意這里， 入隊操作成功之后，此時隊列非空， 則喚醒notEmpty隊列中的節點
notEmpty.signal();
}
take方法
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 可響應中斷地獲取鎖
lock.lockInterruptibly();
try {
// 如果隊列為空， notWait陷入阻塞，直到被signal
while (count == 0)
notEmpty.await();
// 出隊操作
return dequeue();
} finally {
// 解鎖
lock.unlock();
}
}
// 出隊操作
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
// 注意這里， 出隊成功之后， 隊列非滿， 則喚醒notFull中的節點
notFull.signal();
return x;
}
Condition的await()方法會將線程包裝為等待節點，加入等待隊列中，並將AQS同步隊列中的節點移除，接着不斷檢查isOnSyncQueue(Node node)，如果在等待隊列中，就一直等着，如果signal將它移到AQS隊列中，則退出循環。
Condition的signal()方法則是先檢查當前線程是否獲取了鎖，接着將等待隊列中的節點通過Node的操作直接加入AQS隊列。線程並不會立即獲取到資源，從while循環退出后，會通過acquireQueued方法加入獲取同步狀態的競爭中。
而上述描述的線程等待or阻塞則是通過LockSupport的park和unpark方法具體實現，具體可以參考AQS和LockSupport相關內容：
參考閱讀
-
方騰飛《Java並發編程的藝術》
-
《Java並發編程之美》