多线程系列(十三) -一文带你搞懂壅闭队列

一、摘要

在之前的文章中，我们介绍了生产者和消耗者模型的最基本实现思路，相信大家对它已经有一个初步的熟悉。
在 Java 的并发包内里另有一个非常重要的接口：BlockingQueue。
BlockingQueue是一个壅闭队列，更为正确的解释是：BlockingQueue是一个基于壅闭机制实现的线程安全的队列。通过它也可以实现生产者和消耗者模型，而且效率更高、安全可靠，相比之前介绍的生产者和消耗者模型，它可以同时实现生产者和消耗者并行运行。

那什么是壅闭队列呢？
简朴的说，就是当参数在入队和出队时，通过加锁的方式来制止线程并发操作时导致的数据异常问题。
在 Java 中，能对线程并发执行进行加锁的方式主要有synchronized和ReentrantLock，其中BlockingQueue接纳的是ReentrantLock方式实现。
与此对应的另有非壅闭机制的队列，主要是接纳 CAS 方式来控制并发操作，例如：ConcurrentLinkedQueue，这个我们在后面的文章再进行分享介绍。
今天我们主要介绍BlockingQueue相关的知识和用法，废话不多说了，进入正题！
二、BlockingQueue 方法介绍

打开BlockingQueue的源码，你会发现它继承自Queue，正如上文提到的，它本质是一个队列接口。

1. public interface BlockingQueue<E> extends Queue<E> {
2.         //...省略
3. }

复制代码

关于队列，我们在之前的集合系列文章中对此有过深入的介绍，本篇就再次简朴的介绍一下。
队列实在是一个数据结构，元素遵循先辈先出的原则，所有新元素的插入，也被称为入队操作，会插入到队列的尾部；元素的移除，也被称为出队操作，会从队列的头部开始移除，从而保证先辈先出的原则。
在Queue接口中，总共有 6 个方法，可以分为 3 类，分别是：插入、移除、查询，内容如下：
方法描述add(e)插入元素，如果插入失败，就抛异常offer(e)插入元素，如果插入成功，就返回 true；反之 falseremove()移除元素，如果移除失败，就抛异常poll()移除元素，如果移除成功，返回 true；反之 falseelement()获取队首元素，如果获取效果为空，就抛异常peek()获取队首元素，如果获取效果为空，返回空对象由于BlockingQueue是Queue的子接口，相识Queue接口内里的方法，有助于我们对BlockingQueue的理解。
除此之外，BlockingQueue还单独扩展了一些特有的方法，内容如下：
方法描述put(e)插入元素，如果没有插入成功，线程会一直壅闭，直到队列中有空间再继续offer(e, time, unit)插入元素，如果在指定的时间内没有插入成功，就返回 false；反之 truetake()移除元素，如果没有移除成功，线程会一直壅闭，直到队列中新的数据被加入poll(time, unit)移除元素，如果在指定的时间内没有移除成功，就返回 false；反之 truedrainTo(Collection c, int maxElements)一次性取走队列中的数据到 c 中，可以指定取的个数。该方法可以提升获取数据效率，不需要多次分批加锁或释放锁分析源码，你会发现相比平凡的Queue子类，BlockingQueue子类主要有以下几个明显的不同点：

• 1.元素插入和移除时线程安全：主要是通过在入队和出队时进行加锁，保证了队列线程安全，加锁逻辑接纳ReentrantLock实现
  
• 2.支持壅闭的入队和出队方法：当队列满时，会壅闭入队的线程，直到队列不满；当队列为空时，会壅闭出队的线程，直到队列中有元素；同时支持超时机制，防止线程一直壅闭
  

三、BlockingQueue 用法详解

打开源码，BlockingQueue接口的实现类非常多，我们重点讲解一下其中的 5 个非常重要的实现类，分别如下表所示。
实现类功能ArrayBlockingQueue基于数组的壅闭队列，使用数组存储数据，需要指定长度，所以是一个有界队列LinkedBlockingQueue基于链表的壅闭队列，使用链表存储数据，默认是一个无界队列；也可以通过构造方法中的capacity设置最大元素数量，所以也可以作为有界队列SynchronousQueue一种没有缓冲的队列，生产者产生的数据直接会被消耗者获取而且立刻消耗PriorityBlockingQueue基于优先级别的壅闭队列，底层基于数组实现，是一个无界队列DelayQueue延迟队列，其中的元素只有到了其指定的延迟时间，才华够从队列中出队下面我们对以上实现类的用法，进行一一介绍。
3.1、ArrayBlockingQueue

ArrayBlockingQueue是一个基于数组的壅闭队列，初始化的时候必须指定队列大小，源码实现比较简朴，接纳的是ReentrantLock和Condition实现生产者和消耗者模型，部分焦点源码如下：

1. public class ArrayBlockingQueue<E> extends AbstractQueue<E>
2.         implements BlockingQueue<E>, java.io.Serializable {
4.         /** 使用数组存储队列中的元素 */
5.         final Object[] items;
7.         /** 使用独占锁ReetrantLock */
8.         final ReentrantLock lock;
10.         /** 等待出队的条件 */
11.         private final Condition notEmpty;
13.         /** 等待入队的条件 */
14.         private final Condition notFull;
16.         /** 初始化时，需要指定队列大小 */
17.         public ArrayBlockingQueue(int capacity) {
18.         this(capacity, false);
19.     }
21.     /** 初始化时，也指出指定是否为公平锁， */
22.     public ArrayBlockingQueue(int capacity, boolean fair) {
23.         if (capacity <= 0)
24.             throw new IllegalArgumentException();
25.         this.items = new Object[capacity];
26.         lock = new ReentrantLock(fair);
27.         notEmpty = lock.newCondition();
28.         notFull =  lock.newCondition();
29.     }
31.     /**入队操作*/
32.     public void put(E e) throws InterruptedException {
33.         checkNotNull(e);
34.         final ReentrantLock lock = this.lock;
35.         lock.lockInterruptibly();
36.         try {
37.             while (count == items.length)
38.                 notFull.await();
39.             enqueue(e);
40.         } finally {
41.             lock.unlock();
42.         }
43.     }
45.     /**出队操作*/
46.     public E take() throws InterruptedException {
47.         final ReentrantLock lock = this.lock;
48.         lock.lockInterruptibly();
49.         try {
50.             while (count == 0)
51.                 notEmpty.await();
52.             return dequeue();
53.         } finally {
54.             lock.unlock();
55.         }
56.     }
57. }

复制代码

把最上面的样例Container中的壅闭队列实现类换成LinkedBlockingQueue，调整如下：

1. public class Container {
3.     /**
4.      * 初始化阻塞队列
5.      */
6.     private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
8.     /**
9.      * 添加数据到阻塞队列
10.      * @param value
11.      */
12.     public void add(Integer value) {
13.         try {
14.             queue.put(value);
15.             System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value);
16.         } catch (InterruptedException e) {
17.             e.printStackTrace();
18.         }
19.     }
21.     /**
22.      * 从阻塞队列获取数据
23.      */
24.     public void get() {
25.         try {
26.             Integer value = queue.take();
27.             System.out.println("消费者："+ Thread.currentThread().getName()+"，value：" + value);
28.         } catch (InterruptedException e) {
29.             e.printStackTrace();
30.         }
31.     }
33. }

复制代码

再次运行效果如下：

1. /**
2. * 生产者
3. */
4. public class Producer extends Thread {
6.     private Container container;
8.     public Producer(Container container) {
9.         this.container = container;
10.     }
12.     @Override
13.     public void run() {
14.         for (int i = 0; i < 6; i++) {
15.             container.add(i);
16.         }
17.     }
18. }

复制代码

可以很清楚的看到，生产者线程和消耗者线程，瓜代并行执行。
3.3、SynchronousQueue

SynchronousQueue是一个没有缓冲的队列，生产者产生的数据直接会被消耗者获取而且立刻消耗，相当于传统的一个请求对应一个应答模式。
相比ArrayBlockingQueue和LinkedBlockingQueue，SynchronousQueue实现机制也不同，它主要接纳队列和栈来实现数据的通报，中间不存储任何数据，生产的数据必须得消耗者处理，线程壅闭方式接纳 JDK 提供的LockSupport park/unpark函数来完成，也支持公平和非公平两种模式。

• 当接纳公平模式时：使用一个 FIFO 队列来管理多余的生产者和消耗者
  
• 当接纳非公平模式时：使用一个 LIFO 栈来管理多余的生产者和消耗者，这也是SynchronousQueue默认的模式
  

部分焦点源码如下：

1. /**
2. * 消费者
3. */
4. public class Consumer extends Thread {
6.     private Container container;
8.     public Consumer(Container container) {
9.         this.container = container;
10.     }
12.     @Override
13.     public void run() {
14.         for (int i = 0; i < 6; i++) {
15.             container.get();
16.         }
17.     }
18. }

复制代码

同样的，把最上面的样例Container中的壅闭队列实现类换成SynchronousQueue，代码如下：

1. /**
2. * 测试类
3. */
4. public class MyThreadTest {
6.     public static void main(String[] args) {
7.         Container container = new Container();
9.         Producer producer = new Producer(container);
10.         Consumer consumer = new Consumer(container);
12.         producer.start();
13.         consumer.start();
14.     }
15. }

复制代码

再次运行效果如下：

1. 生产者：Thread-0，add：0
2. 生产者：Thread-0，add：1
3. 生产者：Thread-0，add：2
4. 生产者：Thread-0，add：3
5. 生产者：Thread-0，add：4
6. 生产者：Thread-0，add：5
7. 消费者：Thread-1，value：0
8. 消费者：Thread-1，value：1
9. 消费者：Thread-1，value：2
10. 消费者：Thread-1，value：3
11. 消费者：Thread-1，value：4
12. 消费者：Thread-1，value：5

复制代码

可以很清楚的看到，生产者线程和消耗者线程，瓜代串行执行，生产者每投递一条数据，消耗者处理一条数据。
3.4、PriorityBlockingQueue

PriorityBlockingQueue是一个基于优先级别的壅闭队列，底层基于数组实现，可以以为是一个无界队列。
PriorityBlockingQueue与ArrayBlockingQueue的实现逻辑，基本相似，也是接纳ReentrantLock来实现加锁的操作。
最大不同点在于：

• 1.PriorityBlockingQueue内部基于数组实现的最小二叉堆算法，可以对队列中的元素进行排序，插入队列的元素需要实现Comparator或者Comparable接口，以便对元素进行排序
  
• 2.其次，队列的长度是可扩展的，不需要显式指定长度，上限为Integer.MAX_VALUE - 8
  

部分焦点源码如下：
[code]public class PriorityBlockingQueue extends AbstractQueue    implements BlockingQueue, java.io.Serializable {         /**队列元素*/    private transient Object[] queue;    /**比较器*/    private transient Comparator