并發王者課-鉑金05：致勝良器-無處不在的“阻塞佇列”究竟是何面目

歡迎來到《[并發王者課](https://juejin.cn/post/6967277362455150628)》，本文是該系列文章中的**第18篇**，

在執行緒的同步中，**阻塞佇列**是一個繞不過去的話題，它是同步器底層的關鍵，所以，我們在本文中將為你介紹阻塞佇列的基本原理，以了解它的作業機制和它在Java中的實作，本文稍微有點長，建議先了解大綱再細看章節，

## 一、阻塞佇列介紹

在生活中，相信你一定見過下圖的人山人海，也見過其中的秩序井然，**混亂，是失控的開始**，想想看，在沒有秩序的情況下，擁擠的人流蜂擁而上十分危險，輕則擠出一身臭汗，重則造成踩踏事故，**而秩序，則讓情況免于混亂，排好隊大家都舒服**，


**面對人流，我們通過排隊解決混亂，而面對多執行緒，我們也通過佇列讓執行緒間免于混亂，這就是阻塞佇列為何而存在，**

所謂阻塞佇列，你可以理解它是這樣的一種佇列：
* **當執行緒試著往佇列里放資料時，如果它已經滿了，那么執行緒將進入等待**；
* **而當執行緒試著從佇列里取資料時，如果它已經空了，那么執行緒將進入等待**，

下面這張圖展示了多執行緒是如何通過阻塞佇列進行協作的：

從圖中可以看到，對于阻塞佇列資料的讀寫并不局限于單個執行緒，往往存在多個執行緒的競爭，

## 二、實作簡單的阻塞佇列

接下來我們先拋開JUC中復雜的阻塞佇列，來設計一個簡單的阻塞佇列，以了解它的核心思想，

在下面的阻塞佇列中，我們設計一個佇列`queue`，并通過`limit`欄位限定它的容量，`enqueue()`方法用于向佇列中放入資料，如果佇列已滿則等待；而`dequeue()`方法則用于從資料中取出資料，如果佇列為空則等待，

```java
public class BlockingQueue {
private final List<Object> queue = new LinkedList<>();
private final int limit;

public BlockingQueue(int limit) {
this.limit = limit;
}

public synchronized void enqueue(Object item) throws InterruptedException {
while (this.queue.size() == this.limit) {
print("佇列已滿，等待中...");
wait();
}
this.queue.add(item);
if (this.queue.size() == 1) {
notifyAll();
}
print(item, "已經放入！");
}

public synchronized Object dequeue() throws InterruptedException {
while (this.queue.size() == 0) {
print("佇列空的，等待中...");
wait();
}
if (this.queue.size() == this.limit) {
notifyAll();
}
Object item = this.queue.get(0);
print(item, "已經拿到！");
return this.queue.remove(0);
}

public static void print(Object... args) {
StringBuilder message = new StringBuilder(getThreadName() + ":");
for (Object arg : args) {
message.append(arg);
}
System.out.println(message);
}

public static String getThreadName() {
return Thread.currentThread().getName();
}
}
```

定義`lanLingWang`執行緒向佇列中放入資料，`niumo`執行緒從佇列中取出資料，

```java
public static void main(String[] args) {
BlockingQueue blockingQueue = new BlockingQueue(1);
Thread lanLingWang = new Thread(() -> {
try {
String[] items = { "A", "B", "C", "D", "E" };
for (String item: items) {
Thread.sleep(500);
blockingQueue.enqueue(item);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
lanLingWang.setName("蘭陵王");
Thread niumo = new Thread(() -> {
try {
while (true) {
blockingQueue.dequeue();
Thread.sleep(1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
lanLingWang.setName("蘭陵王");
niumo.setName("牛魔王");

lanLingWang.start();
niumo.start();
}
```

運行結果如下：

```shell
牛魔王:佇列空的，等待中...
蘭陵王:A已經放入！
牛魔王:A已經拿到！
蘭陵王:B已經放入！
牛魔王:B已經拿到！
蘭陵王:C已經放入！
蘭陵王:佇列已滿，等待中...
牛魔王:C已經拿到！
蘭陵王:D已經放入！
蘭陵王:佇列已滿，等待中...
牛魔王:D已經拿到！
蘭陵王:E已經放入！
牛魔王:E已經拿到！
牛魔王:佇列空的，等待中...

```

從結果中可以看到，設計的阻塞佇列已經可以有效作業，你可以仔細地品一品輸出的結果，當然，這個阻塞是極其簡單的，在下面一節中，我們將介紹Java中的阻塞佇列設計，

## 三、Java中的BlockingQueue

Java中的阻塞佇列有兩個核心介面：**BlockingQueue**和**BlockingDeque**，相關的介面實作設繼承關系如下圖所示，相比于上一節中我們自定義的阻塞佇列，Java中的實作要復雜很多，不過，你不必為此擔心，**理解阻塞佇列最重要的是理解它的思想和實作的思路，況且Java中的實作其實很有意思，讀起來也比較輕松**，

從圖中可以看出，BlockingQueue介面繼承了**Queue**介面和**Collection**介面，并有LinkedBlockingQueue和ArrayBlockingQueue兩種實作，這里有個有意思的地方，**繼承**Queue**介面很容易理解，可以為什么要繼承**Collection**介面？先賣個關子，你可以思考一會，稍后會給出答案**，


### 1. 核心方法

BlockingQueue中義了關于阻塞佇列所需要的一系列方法，它們彼此之間看起來很像，從表面上看不出明顯的差別，對于這些方法，你不必死記硬背，下圖的表格中將這些方法分為了**A、B、C、D**這四種型別，分類之后再去理解它們會容易很多：

|型別|A 拋出例外|B 回傳特定值| C 阻塞|D 超時限定|
|---|---|---|---|---|
| Insert |` add(e)` | `offer(e)` | `put(e)` | `offer(e, time, unit)` |
| Remove | `remove()` | `poll()` | `take(`) | `poll(time, unit)` |
| Examine | `Element()` | `peek()` | -- | -- |

其中部分關鍵方法的解釋如下：

* `add(E e)`：在不違反容量限制的前提下，向佇列中插入資料，**如果成功，回傳true，否則拋出例外**；
* `offer(E e)`：在不違反容量限制的前提下，向佇列中插入資料，**如果成功，回傳`true`，否則回傳`false`**；
* `offer(E e, long timeout, TimeUnit unit)`：如果佇列中沒有足夠的空間，將等待一段時間；
* `put(E e)`：在不違反容量限制的前提下，向佇列中插入資料，**如果沒有足夠的空間，將進入等待**；
* `poll(long timeout, TimeUnit unit)`：從佇列的頭部獲取資料，并移除資料，如果沒有資料的話，將會等待指定的時間；
* `take()`：從佇列的頭部獲取資料并移除，如果沒有可用資料，將進入等待

將這些方法填入前面的那張圖，它應該長這樣：

### 2. LinkedBlockingQueue

LinkedBlockingQueue實作了BlockingQueue介面，遵從先進先出（FIFO）的原則，提供了可選的有界阻塞佇列（ Optionally Bounded ）的能力，并且是執行緒安全的，

* **核心資料結構**
* `int capacity`: 設定佇列容量；
* `Node<E> head`: 佇列的頭部元素；
* `Node<E> last`: 佇列的尾部元素；
* `AtomicInteger count`: 佇列中元素的總數統計，

LinkedBlockingQueue的資料結構并不復雜，不過需要注意的是，資料結構中并不包含List，僅有`head`和`last`兩個Node，設計上比較巧妙，

* **核心構造**
* `LinkedBlockingQueue()`: 空構造；
* `LinkedBlockingQueue(int capacity)`: 指定容量構造，
* **執行緒安全性**
* `ReentrantLock takeLock`: 獲取元素時的鎖；
* `ReentrantLock putLock`: 寫入元素時的鎖，

**注意，LinkedBlockingQueue有兩把鎖，讀取和寫入的鎖是分離的**！這和下面的ArrayBlockingQueue并不相同，

下面截取了LinkedBlockingQueue中讀寫的部分代碼，值得你仔細品一品，**品的時候，要重點關注兩把鎖的使用和讀寫時資料結構是如何變化的**，

* **佇列插入示例代碼分析**

```java
public boolean add(E e) {
addLast(e);
return true;
}

public void addLast(E e) {
if (!offerLast(e))
throw new IllegalStateException("Deque full");
}

public boolean offerFirst(E e) {
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
return linkFirst(node);
} finally {
lock.unlock();
}
}
```
* **佇列讀取示例代碼分析**

```java
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
return pollFirst(timeout, unit);
}
public E pollFirst(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
E x;
while ( (x = unlinkFirst()) == null) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
return x;
} finally {
lock.unlock();
}
}
```

最后說下LinkedBlockingQueue為什么要繼承Collection介面，我們知道，Collection介面有`remove()`這樣的移除方法，而這些方法在佇列中也是有使用場景的，比如，你把一個資料錯誤地放入了佇列，或者你需要移除已經失效的資料，那么Collection的一些方法就派上了用場，

### 3. ArrayBlockingQueue

ArrayBlockingQueue是BlockingQueue介面的另外一種實作，**它與LinkedBlockingQueue在設計目標上的的關鍵不同，在于它是有界的**，

* **核心資料結構**

* `Object[] items`: 佇列元素集合；
* `int takeIndex`: 下次獲取資料時的索引位置；
* `int putIndex`: 下次寫入資料時的索引位置；
* `int count`: 佇列總量計數，

從資料結構中可以看出，ArrayBlockingQueue使用的是陣列，而陣列是有界的，

* **核心構造**
* `ArrayBlockingQueue(int capacity)`： 限定容量的構造；
* `ArrayBlockingQueue(int capacity, boolean fair)`： 限定容量和公平性，默認是不公平的；
* `ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)`：帶有初始化佇列元素的構造，

* **執行緒安全性**

* `ReentrantLock lock`：佇列讀取和寫入的鎖，

在讀寫鎖方面，前面已經說過，LinkedBlockingQueue和ArrayBlockingQueue是不同的，ArrayBlockingQueue只有一把鎖，讀寫用的都是它，

* **佇列寫入示例代碼分析**

```java
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} 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.signal();
}
```

下面截取了ArrayBlockingQueue中讀寫的部分代碼，值得你仔細品一品，**品的時候，要重點關注讀寫鎖的使用和讀寫時資料結構是如何變化的**，

* **佇列讀取示例代碼分析**

```java
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
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.signal();
return x;
}
```

## 四、Java中的BlockingDeque

在Java中，BlockingDeque與BlockingQueue是一對孿生兄弟似的存在，它們長得實在太像了，不注意的話很容易混淆，

**但是，BlockingDeque與BlockingQueue核心不同在于，BlockingQueue只能夠從尾部寫入、從頭部讀取，使用上很有限制，而BlockingDeque則支持從任意端讀寫，在讀寫時可以指定頭部和尾部，豐富了阻塞佇列的使用場景**，

### 1. 核心方法

相較于BlockingQueue，BlockingDeque的方法顯然要更豐富一些，畢竟它支持了雙端的讀寫，**但是，豐富歸豐富，在型別上仍然和BlockingQueue是一致的，你仍然可以參考上面的A、B、C、D四種型別來分類理解**，為了節約篇幅，我們這里就不再羅列，只選取了其中的部分方法作了解釋：

* `add(E e)`：在不違反容量限制的前提下，在對列的尾部插入資料；
* `addFirst(E e)`：從頭部插入資料，容量不夠就拋錯；
* `addLast(E e)`：從尾部插入資料，容量不夠就拋錯；
* `getFirst()`：從頭部讀取資料；
* `getLast()`：從尾部讀取資料，但不會移除資料；
* `offer(E e)`：寫入資料；
* `offerFirst(E e)`：從頭部寫入資料，

將BlockingDeue放入前面的那張圖，就是這樣：


### 2. LinkedBlockingDeue

LinkedBlockingDeue是BlockingDeque的核心實作，

* **核心資料結構**
* `int capacity`：容量設定；
* `Node<E> head`：佇列頭部；
* `Node<E> last`：佇列尾部；
* `int count`：佇列計數，
* **核心構造**
* `LinkedBlockingDeque()`： 空的構造；
* `LinkedBlockingDeque(int capacity)`： 指定容量的構造；
* `LinkedBlockingDeque(Collection<? extends E> c)`：構造時初始化佇列，
* **執行緒安全性**

* `ReentrantLock lock`：讀寫鎖，**注意，讀寫用的是同一把鎖**，

下面截取了LinkedBlockingDeue中讀寫的部分代碼，值得你仔細品一品，**品的時候，要重點關注讀寫鎖的使用和讀寫時資料結構是如何變化的**

* **佇列插入示例代碼分析**

```java
public void addFirst(E e) {
if (!offerFirst(e))
throw new IllegalStateException("Deque full");
}
public boolean offerFirst(E e) {
if (e == null) throw new NullPointerException();
Node < E > node = new Node < E > (e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
return linkFirst(node);
} finally {
lock.unlock();
}
}
```

* **佇列讀取示例代碼分析**

```java
public E pollFirst() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return unlinkFirst();
} finally {
lock.unlock();
}
}

```
## 小結

以上就是關于阻塞佇列的全部內容，相較于前面的系列文章，這次的內容明顯增加了很多，**看起來很簡單，但是不要小瞧它**，理解阻塞佇列，首先要理解它所要解決的問題，以及它的介面設計，**介面的設計往往表示的是它所提供的核心能力，所以理解了介面的設計，就成功了一半**，

在Java中，從介面層面，阻塞佇列分為BlockingQueue和BlockingDeque的兩大類，其主要差異在于雙端讀寫的限制不同，其中，BlockingQueue有LinkedBlockingDeue和ArrayBlockingQueue兩種關鍵實作，而BlockingDeque則有LinkedBlockingDeue實作，

正文到此結束，恭喜你又上了一顆星?

**夫子的試煉**

* 從資料機構、佇列的初始化、鎖、性能等方面比較LinkedBlockingDeue和ArrayBlockingQueue的不同，

**延伸閱讀與參考資料**

* [Talk about LinkedBlockingQueue](https://programmer.help/blogs/talk-about-linkedblockingqueue.html)
* [Blocking Queues](http://tutorials.jenkov.com/java-concurrency/blocking-queues.html)
* [《并發王者課》大綱與更新進度總覽](https://juejin.cn/post/6967277362455150628)

**關于作者**

關注公眾號【**技術八點半**】，及時獲取文章更新，傳遞有品質的技術文章，記錄平凡人的成長故事，偶爾也聊聊生活和理想，早晨8：30推送作者品質原創，晚上20：30推送行業深度好文，

如果本文對你有幫助，歡迎**點贊**、**關注**、**監督**，我們一起**從青銅到王者**，

標籤：Java

上一篇：Netty 框架學習 —— 編解碼器框架

下一篇：用了很多年的 CMS 垃圾收集器，終于換成了 G1，真香！！