欢迎来到《[并发王者课](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推送行业深度好文。
如果本文对你有帮助，欢迎**点赞**、**关注**、**监督**，我们一起**从青铜到王者**。