✨ 『并发包入坑指北』之阻塞队列

crossoverJie · crossoverJie · commit 5bd9b3b4df3a · 2019-04-15T11:00:04.000+08:00
diff --git a/README.md b/README.md
@@ -44,6 +44,7 @@
 - [深入理解线程通信](https://github.com/crossoverJie/JCSprout/blob/master/MD/concurrent/thread-communication.md)
 - [一个线程罢工的诡异事件](docs/thread/thread-gone.md)
 - [线程池中你不容错过的一些细节](docs/thread/thread-gone2.md)
+- [『并发包入坑指北』之阻塞队列](docs/thread/ArrayBlockingQueue.md)
 
 ### JVM
 - [Java 运行时内存划分](https://github.com/crossoverJie/JCSprout/blob/master/MD/MemoryAllocation.md)
diff --git a/docs/_sidebar.md b/docs/_sidebar.md
@@ -18,6 +18,7 @@
   - [深入理解线程通信](thread/thread-communication.md)
   - [一个线程罢工的诡异事件](thread/thread-gone.md)
   - [线程池中你不容错过的一些细节](thread/thread-gone2.md)
+  - [『并发包入坑指北』之阻塞队列](thread/ArrayBlockingQueue.md)
 
 - JVM
 
diff --git a/docs/thread/ArrayBlockingQueue.md b/docs/thread/ArrayBlockingQueue.md
@@ -0,0 +1,283 @@
+![](https://ws4.sinaimg.cn/large/006tNc79ly1g1vn9xpgp4j31ak0u013m.jpg)
+
+
+
+# 前言
+
+较长一段时间以来我都发现不少开发者对 jdk 中的 `J.U.C`（java.util.concurrent）也就是 Java 并发包的使用甚少，更别谈对它的理解了；但这却也是我们进阶的必备关卡。
+
+之前或多或少也分享过相关内容，但都不成体系；于是便想整理一套与并发包相关的系列文章。
+
+其中的内容主要包含以下几个部分：
+
+- 根据定义自己实现一个并发工具。
+- JDK 的标准实现。
+- 实践案例。
+
+
+基于这三点我相信大家对这部分内容不至于一问三不知。
+
+既然开了一个新坑，就不想做的太差；所以我打算将这个列表下的大部分类都讲到。
+
+![](https://ws2.sinaimg.cn/large/006tNc79ly1g1vpwdqbkrj30ab09nmy9.jpg)
+
+
+所以本次重点讨论 `ArrayBlockingQueue`。
+
+# 自己实现
+
+在自己实现之前先搞清楚阻塞队列的几个特点：
+
+- 基本队列特性：先进先出。
+- 写入队列空间不可用时会阻塞。
+- 获取队列数据时当队列为空时将阻塞。
+
+
+实现队列的方式多种，总的来说就是数组和链表；其实我们只需要搞清楚其中一个即可，不同的特性主要表现为数组和链表的区别。
+
+这里的 `ArrayBlockingQueue` 看名字很明显是由数组实现。
+
+我们先根据它这三个特性尝试自己实现试试。
+
+## 初始化队列
+
+我这里自定义了一个类：`ArrayQueue`，它的构造函数如下：
+
+```java
+    public ArrayQueue(int size) {
+        items = new Object[size];
+    }
+```
+
+很明显这里的 `items` 就是存放数据的数组；在初始化时需要根据大小创建数组。
+
+![](https://ws1.sinaimg.cn/large/006tNc79ly1g1wd71n229j30wb0u043w.jpg)
+
+## 写入队列
+
+写入队列比较简单，只需要依次把数据存放到这个数组中即可，如下图：
+
+![](https://ws3.sinaimg.cn/large/006tNc79ly1g1we7yeykej30b0060mxc.jpg)
+
+但还是有几个需要注意的点：
+
+- 队列满的时候，写入的线程需要被阻塞。
+- 写入过队列的数量大于队列大小时需要从第一个下标开始写。
+
+先看第一个`队列满的时候，写入的线程需要被阻塞`，先来考虑下如何才能使一个线程被**阻塞**，看起来的表象线程卡住啥事也做不了。
+
+有几种方案可以实现这个效果:
+
+- `Thread.sleep(timeout)`线程休眠。
+- `object.wait()` 让线程进入 `waiting` 状态。
+
+> 当然还有一些 `join、LockSupport.part` 等不在本次的讨论范围。
+
+阻塞队列还有一个非常重要的特性是：当队列空间可用时（取出队列），写入线程需要被唤醒让数据可以写入进去。
+
+所以很明显`Thread.sleep(timeout)`不合适，它在到达超时时间之后便会继续运行；达不到**空间可用时**才唤醒继续运行这个特点。
+
+其实这样的一个特点很容易让我们想到 Java 的等待通知机制来实现线程间通信；更多线程见通信的方案可以参考这里：[深入理解线程通信](https://crossoverjie.top/2018/03/16/java-senior/thread-communication/#%E7%AD%89%E5%BE%85%E9%80%9A%E7%9F%A5%E6%9C%BA%E5%88%B6)
+
+所以我这里的做法是，一旦队列满时就将写入线程调用 `object.wait()` 进入 `waiting` 状态，直到空间可用时再进行唤醒。
+
+```java
+    /**
+     * 队列满时的阻塞锁
+     */
+    private Object full = new Object();
+
+    /**
+     * 队列空时的阻塞锁
+     */
+    private Object empty = new Object();
+```
+
+![](https://ws2.sinaimg.cn/large/006tNc79ly1g1wf8de8jzj30te0tin1i.jpg)
+
+所以这里声明了两个对象用于队列满、空情况下的互相通知作用。
+
+
+在写入数据成功后需要使用 `empty.notify()`，这样的目的是当获取队列为空时，一旦写入数据成功就可以把消费队列的线程唤醒。
+
+
+> 这里的 wait 和 notify 操作都需要对各自的对象使用 `synchronized` 方法块，这是因为 wait 和 notify 都需要获取到各自的锁。
+
+## 消费队列
+
+上文也提到了：当队列为空时，获取队列的线程需要被阻塞，直到队列中有数据时才被唤醒。
+
+![](https://ws1.sinaimg.cn/large/006tNc79ly1g1wfhr3r6qj30tg0tiwit.jpg)
+
+代码和写入的非常类似，也很好理解；只是这里的等待、唤醒恰好是相反的，通过下面这张图可以很好理解：
+
+![](https://ws3.sinaimg.cn/large/006tNc79ly1g1wfwr016gj30o20ksq59.jpg)
+
+总的来说就是：
+
+- 写入队列满时会阻塞直到获取线程消费了队列数据后唤醒**写入线程**。
+- 消费队列空时会阻塞直到写入线程写入了队列数据后唤醒**消费线程**。
+
+
+## 测试
+
+先来一个基本的测试：单线程的写入和消费。
+
+![](https://ws2.sinaimg.cn/large/006tNc79ly1g1wg97uqgpj30uu0dqwgu.jpg)
+
+```log
+3
+123
+1234
+12345
+```
+
+通过结果来看没什么问题。
+
+---
+
+当写入的数据超过队列的大小时，就只能消费之后才能接着写入。
+
+![](https://ws3.sinaimg.cn/large/006tNc79ly1g1wgmshqfyj316o0n2ae5.jpg)
+
+```log
+2019-04-09 16:24:41.040 [Thread-0] INFO  c.c.concurrent.ArrayQueueTest - [Thread-0]123
+2019-04-09 16:24:41.040 [main] INFO  c.c.concurrent.ArrayQueueTest - size=3
+2019-04-09 16:24:41.047 [main] INFO  c.c.concurrent.ArrayQueueTest - 1234
+2019-04-09 16:24:41.048 [main] INFO  c.c.concurrent.ArrayQueueTest - 12345
+2019-04-09 16:24:41.048 [main] INFO  c.c.concurrent.ArrayQueueTest - 123456
+```
+
+从运行结果也能看出只有当消费数据后才能接着往队列里写入数据。
+
+---
+
+![](https://ws3.sinaimg.cn/large/006tNc79ly1g1wiskvki8j30yy0eo0ve.jpg)
+
+![](https://ws2.sinaimg.cn/large/006tNc79ly1g1witm4twpj31q60ai0vz.jpg)
+
+而当没有消费时，再往队列里写数据则会导致写入线程被阻塞。
+
+
+
+### 并发测试
+
+![](https://ws3.sinaimg.cn/large/006tNc79ly1g1wiwyz4j5j30vz0u044f.jpg)
+
+三个线程并发写入300条数据，其中一个线程消费一条。
+
+```log
+=====0
+299
+```
+
+最终的队列大小为 299，可见线程也是安全的。
+
+> 由于不管是写入还是获取方法里的操作都需要获取锁才能操作，所以整个队列是线程安全的。
+
+
+# ArrayBlockingQueue
+
+下面来看看 JDK 标准的 `ArrayBlockingQueue` 的实现，有了上面的基础会更好理解。
+
+## 初始化队列
+
+![](https://ws1.sinaimg.cn/large/006tNc79ly1g1wkaau8w7j30ze0lcagb.jpg)
+
+看似要复杂些，但其实逐步拆分后也很好理解：
+
+第一步其实和我们自己写的一样，初始化一个队列大小的数组。
+
+
+第二步初始化了一个重入锁，这里其实就和我们之前使用的 `synchronized` 作用一致的；
+
+只是这里在初始化重入锁的时候默认是`非公平锁`，当然也可以指定为 `true` 使用公平锁；这样就会按照队列的顺序进行写入和消费。
+
+> 更多关于 `ReentrantLock` 的使用和原理请参考这里：[ReentrantLock 实现原理](https://crossoverjie.top/2018/01/25/ReentrantLock/)
+
+三四两步则是创建了 `notEmpty notFull` 这两个条件，他的作用于用法和之前使用的 `object.wait/notify` 类似。
+
+这就是整个初始化的内容，其实和我们自己实现的非常类似。
+
+
+## 写入队列
+
+![](https://ws2.sinaimg.cn/large/006tNc79ly1g1wktuhxzuj30tk0bqq55.jpg)
+![](https://ws2.sinaimg.cn/large/006tNc79ly1g1wktfkwu2j30ug09ugnn.jpg)
+
+其实会发现阻塞写入的原理都是差不多的，只是这里使用的是 Lock 来显式获取和释放锁。
+
+同时其中的 `notFull.await();notEmpty.signal();` 和我们之前使用的 `object.wait/notify` 的用法和作用也是一样的。
+
+
+当然它还是实现了超时阻塞的 `API`。
+
+![](https://ws4.sinaimg.cn/large/006tNc79ly1g1wl1n7ir5j30vm0iqdjb.jpg)
+
+也是比较简单，使用了一个具有超时时间的等待方法。 
+
+## 消费队列
+
+再看消费队列：
+
+![](https://ws1.sinaimg.cn/large/006tNc79ly1g1wl3vcsioj30tc0ayq4y.jpg)
+![](https://ws4.sinaimg.cn/large/006tNc79ly1g1wl4cfrnlj30u00eq0vm.jpg)
+
+也是差不多的，一看就懂。
+
+而其中的超时 API 也是使用了 `notEmpty.awaitNanos(nanos)` 来实现超时返回的，就不具体说了。
+
+
+# 实际案例
+
+说了这么多，来看一个队列的实际案例吧。
+
+背景是这样的：
+
+> 有一个定时任务会按照一定的间隔时间从数据库中读取一批数据，需要对这些数据做校验同时调用一个远程接口。
+
+
+简单的做法就是由这个定时任务的线程去完成读取数据、消息校验、调用接口等整个全流程；但这样会有一个问题：
+
+假设调用外部接口出现了异常、网络不稳导致耗时增加就会造成整个任务的效率降低，因为他都是串行会互相影响。
+
+
+所以我们改进了方案：
+
+![](https://ws1.sinaimg.cn/large/006tNc79ly1g1wm1v7mfxj30qs0aiq4g.jpg)
+
+其实就是一个典型的生产者消费者模型：
+
+- 生产线程从数据库中读取消息丢到队列里。
+- 消费线程从队列里获取数据做业务逻辑。
+
+这样两个线程就可以通过这个队列来进行解耦，互相不影响，同时这个队列也能起到缓冲的作用。
+
+但在使用过程中也有一些小细节值得注意。
+
+因为这个外部接口是支持批量执行的，所以在消费线程取出数据后会在内存中做一个累加，一旦达到阈值或者是累计了一个时间段便将这批累计的数据处理掉。
+
+但由于开发者的大意，在消费的时候使用的是 `queue.take()` 这个阻塞的 API；正常运行没啥问题。
+
+可一旦原始的数据源，也就是 DB 中没数据了，导致队列里的数据也被消费完后这个消费线程便会被阻塞。
+
+这样上一轮积累在内存中的数据便一直没机会使用，直到数据源又有数据了，一旦中间间隔较长时便可能会导致严重的业务异常。
+
+所以我们最好是使用 `queue.poll(timeout)` 这样带超时时间的 api，除非业务上有明确的要求需要阻塞。
+
+这个习惯同样适用于其他场景，比如调用 http、rpc 接口等都需要设置合理的超时时间。
+
+# 总结
+
+关于 `ArrayBlockingQueue` 的相关分享便到此结束，接着会继续更新其他并发容器及并发工具。
+
+对本文有任何相关问题都可以留言讨论。
+
+
+
+本文涉及到的所有源码：
+
+https://github.com/crossoverJie/JCSprout/blob/master/src/main/java/com/crossoverjie/concurrent/ArrayQueue.java
+
+
+**你的点赞与分享是对我最大的支持**
