Java集合--阻塞队列及各种实现的解析

阻塞队列（Blocking Queue）

一、队列的定义

说的阻塞队列，就先了解下什么是队列，队列也是一种特殊的线性表结构，在线性表的基础上加了一条限制：那就是一端入队列，一端出队列，且需要遵循FIFO（先进先出）的原则

队列的入口一端叫做队尾（rear），出口一端叫做队头（front），最简单的比如就是排队买火车票，新加入的排队者必须在队尾插入，而下一个排队结束的永远的队伍的第一个人

如下图示：

二、阻塞队列的定义

了解了队列，就很好理解什么是阻塞队列，既然是阻塞的，那就是需要等待，而队列的操作只有入队和出队，那就是在入队和出队的时候等待了，所以总结下阻塞队列就是：

1、入队时，如果队列已经满了，就阻塞等待直到队列中有位置可以插入

2、出队时，如果队列中为空，就阻塞等待直到队列中有数据可以出队列

三、Java队列的定义

Java集合体系分为Collection和Map两大阵营，而队列Queue则属于Collection系列

3.1、Queue接口

Queue接口继承之Collection接口，在Collection的继承上增加了队列独有的方法

boolean offer（E e）：向队列中插入元素，插入成功返回true；插入失败返回false

E peek()：获取队列头部的元素，如果队列为空则返回null

E poll()：获取并移除头部的元素，队列为空则返回null，和peek的区别是获取头部元素之后将头部元素删除，符合队列的出队操作

E remove()：获取并移除头部的元素，如果为空则抛异常

3.2、BlockingQueue接口

BlockingQueue接口是继承之Queue，在Queue方法的继承上添加了抛出异常操作，因为BlockingQueue上阻塞队列，所以就存在阻塞时间过长需要中断阻塞操作，或者是超时中断阻塞操作

而BlockingQueue也就是阻塞队列的*接口，BlockingQueue不同的实现类就是实现了不同的阻塞队列的效果

阻塞队列的入队和出队操作可以分为多种操作方式：

1、抛异常：入队时队列满了直接抛异常；出队时队列为空直接抛异常  如：add方法和remove方法

2、返回boolean：入队或出队成功返回true；失败返回false 如：offer方法和poll方法

3、阻塞：入队或出队失败则一直阻塞直到成功或者被其他线程唤醒 如：put方法和take方法

4、超时阻塞：入队和出队失败则阻塞指定的时间，超时了还是失败则取消阻塞 如：offer（timeout）和poll（timeout）方法

四、Java中阻塞队列的不同实现

4.1、ArrayBlockingQueue

基于数组实现的阻塞队列，初始化时需要定义数组的大小，也就是队列的大小，所以这个队列是一个有界队列。

ArrayBlockingQueue的主要属性有：

     //存储元素的数组
     final Object[] items;

     //下一个出队的索引
     int takeIndex;

     //下一个入队的索引
     int putIndex;

     //队列的大小
     int count;

     //可重入锁
     final ReentrantLock lock;

     //出队操作的等待
     private final Condition notEmpty;

     //入队操作的等待
     private final Condition notFull;

实现原理：通过可重入锁ReenTrantLock+Condition 来实现多线程之间的同步效果

入队过程：

add方法：插入成功返回true；插入失败抛异常

put方法：插入元素到尾部，如果失败则调用Condition.await()方法进行阻塞等待，直到被唤醒；

offer方法：插入元素到尾部，如果失败则直接返回false，

offer(timeout)：插入元素到尾部，如果失败则调用Condition.await（timeout）方法进行阻塞等待指定时间，直到被唤醒或阻塞超时，还是失败就返回false

而一旦插入成功，就会唤醒出队的等待操作，执行出队的Condition的signal()方法

出队过程：

主要方法为：poll（）、take（）、remove（）

基本上和入队过程类似，出队结束会唤醒入队的等待操作，执行入队的Condition的signal()方法

而不管是入队操作还是出队操作，都会通过ReentrantLock来控制同步效果，通过两个Condition来控制线程之间的通信效果

另外入队和出队操作分别通过两个索引 takeIndex 和putIndex来指定数组的位置，默认从0开始分别递增，如果达到数组的容量大小，就表示到了数组的边界了，此时再设置index=0，相当于数组是一个环形数组

环形数组的好处是增删数据时不需要挪动数组中的其他数据，只需要改变入队和出队的指针即可。而如果不是环形数组而是顺序数组的话，入队和出队就需要大量移动数据，否则数组空间一下就被用完了，性能较差

4.2、LinkedBlockingQueue

基于链表实现的阻塞队列，既然是链表，那么就可以看出这种阻塞队列含有链表的特性，那就是*。但是实际上LinkedBlockingQueue是有界队列，默认大小是Integer的最大值，而也可以通过构造方法传入固定的capacity大小设置

LinkedBlockingQueue有一个内部类Node，属性有：E ite和Node next，所以可以看出LinkedBlockingQueue是一个单向链表

基本属性为：

 //队列大小，默认为Integer的最大值
     private final int capacity;

     //当前队列元素个数
     private final AtomicInteger count = new AtomicInteger();

     //队列的头部元素
     transient LinkedBlockingQueue.Node<E> head;

     //队列的尾部元素
     private transient LinkedBlockingQueue.Node<E> last;

     //出队锁
     private final ReentrantLock takeLock = new ReentrantLock();

     //出队的condition
     private final Condition notEmpty = takeLock.newCondition();

     //入队锁
     private final ReentrantLock putLock = new ReentrantLock();

     //入队的condition
     private final Condition notFull = putLock.newCondition();

可以看出LinkedBlockingQueue的属性和ArrayBlockingQueue的属性大致差不多，都是通过ReentrantLock和Condition来实现多线程之间的同步，而LinkedBlockingQueue却多了一个ReentrantLock，而不是入队和出队共用同一个锁

那么为什么ArrayBlockingQueue只需要一个ReentrantLock而LinkedBlockingQueue需要两个ReentrantLock呢？

个人想法；

首先，ReentrantLock肯定是越多越好，锁越多那么相同锁的竞争就越少；LinkedBlockingQueue分别有入队锁和出队锁，所以入队和出队的时候不会有竞争锁的关系；而ArrayBlockingQueue只有一个Lock，那么不管是入队还是出队，都需要竞争同一个锁，所以效率会低点。ArrayBlockingQueue是环形数组结构，入队的地址和出队的地址可能是同一个，比如数组table大小为1，那么第一次入队和出队需要操作的位置都是table[0]这个元素，所以入队和出队必须共用同一把锁；而LinkedBlockingQueue是链表形式，内存地址是散列的，入队的元素地址和出队的元素地址永远不可能会是同一个地址。所以可以采用两个锁，分别对入队进行加锁同步和对出队进行加锁同步即可。

4.3、DelayQueue

延迟队列,顾名思义就是只有当元素达到指定的时间后才可以从队列中取出。

根据这个思路可以满足下面几种需求：

1.定时任务：将任务放入队列中设置时间，循环阻塞地从队列中取任务，当从队列中取出数据就表示时间到了

2.缓存过期：循环从队列中取数据，一旦取出数据就表示数据过期了，直接删除即可

DelayQueue主要也是通过ReentrantLock+Condition来保证线程安全，而内部还采用了ProrityQueue来保证队列的优先级，实际就是按延时的时间来进行排序，延迟时间最短的排在队列的头部，

所以每次从头部获取的元素都是最先会过期的数据。

4.4、PriorityBlockingQueue

有优先级的阻塞队列，底层也是通过数组实现，默认初始容量为11，容量不够会自动扩容，扩容的最大值为Integer的最大值-8（有些虚拟机再实现数组头部存储内容所预留的空间），所以基本上可以认为是*阻塞队列

扩容时的线程安全通过ReentrantLock+CAS+volatine实现

用法基本上和ArrayBlockingQueue差不多，只不过PriorityBlockingQueue相当于是*，另外最重要的一点是它是有优先级的，既然有优先级就涉及到排序

PriorityBlockingQueue默认采用Comparator，或者存储的元素有自定义的比较器。

而存储数据的数组也不是简单的数组，而是采用了二叉堆的数据结构，同时满足完全二叉树+堆的数据结构（最大堆上层的元素必须大于所有下层的元素；最小堆或者上层的元素必须小于所有下层的元素）

而PriorityBlockingQueue默认是采用的最小堆，即每次取出的元素都是优先级最小的。那么问题来了，如果通过数组来实现一个二叉堆呢？见下面的图解：

先看下二叉堆的数据结构：

可以看出二叉堆的上层永远比下层的优先级要高，而且可以发现上层节点和子节点的关系：父节点序号=（子节点序号-1）/2；左字节点序号=父节点序号*2+1；右字节点序号=父节点序号*2+2

那么上面的数据结构就可以通过数组来存储，如下图示：

插入操作：将元素直接插入到最底层的节点，如上图插入节点88

1. 节点88成为节点40到左子节点，新加入的节点遍历和自己的父节点进行比较，

2.节点88比节点40大，两者互换位置；继续88比80大，互换位置，88比100小，位置不动，

3.结果是新插入的节点88成为节点100的左子节点，节点80成为节点88点右子节点，节点40成功节点80点右子节点

删除操作：阻塞队列的删除是只删除位置为0的元素，也就是节点100

1.直接取出数组[0]位置的元素

2.将队列尾部节点放到头部来，如上图就是把节点5放到最顶层，

3.然后将头节点依次和子节点进行比较，然后进行位置互换操作

所以每次插入和删除元素最多操作次数就是 二叉堆的高度

代码不再分析，主要就是实现来二叉堆的逻辑，并且通过ReentrantLock+Condition来保证线程间的同步效果

4.5、SynchronousQueue

SynchonousQueue是比较特殊的阻塞队列，特殊之处就是这个叫队列的队列没有容量，又或者说容量为0，所以一旦有元素插入此队列，由于没有容量，就必须被阻塞直到元素被取出

所以SynchronousQueue更像是一个通道，一端发数据，一端消费数据，数据不可以被堆积，发送方或消费方处理不过来或者是不处理都会导致阻塞

五、阻塞队列的应用

5.1、线程池

线程池的构造函数就包含了阻塞队列