七种阻塞队列

　　在前面咱们接触的队列都是非阻塞队列，好比PriorityQueue、LinkedList（LinkedList是双向链表，它实现了Dequeue接口）。

　　使用非阻塞队列的时候有一个很大问题就是：它不会对当前线程产生阻塞，那么在面对相似消费者-生产者的模型时，就必须额外地实现同步策略以及线程间唤醒策略，这个实现起来就很是麻烦。可是有了阻塞队列就不同了，它会对当前线程产生阻塞，好比一个线程从一个空的阻塞队列中取元素，此时线程会被阻塞直到阻塞队列中有了元素。当队列中有元素后，被阻塞的线程会自动被唤醒（不须要咱们编写代码去唤醒）。这样提供了极大的方便性。

一. 什么是阻塞队列

　 　阻塞队列（BlockingQueue）是一个支持两个附加操做的队列。这两个附加的操做支持阻塞的插入和移除方法。

　　1）支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不满。

　　2）支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。

　　阻塞队列经常使用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。

1.非阻塞队列中的几个主要方法：

　　add(E e):将元素e插入到队列末尾，若是插入成功，则返回true；若是插入失败（即队列已满），则会抛出异常；

　　remove()：移除队首元素，若移除成功，则返回true；若是移除失败（队列为空），则会抛出异常；

　　offer(E e)：将元素e插入到队列末尾，若是插入成功，则返回true；若是插入失败（即队列已满），则返回false；

　　poll()：移除并获取队首元素，若成功，则返回队首元素；不然返回null；

　　peek()：获取队首元素，若成功，则返回队首元素；不然返回null

　　对于非阻塞队列，通常状况下建议使用offer、poll和peek三个方法，不建议使用add和remove方法。由于使用offer、poll和peek三个方法能够经过返回值判断操做成功与否，而使用add和remove方法却不能达到这样的效果。注意，非阻塞队列中的方法都没有进行同步措施。

2.阻塞队列中的几个主要方法：

　　阻塞队列包括了非阻塞队列中的大部分方法，上面列举的5个方法在阻塞队列中都存在，可是要注意这5个方法在阻塞队列中都进行了同步措施。除此以外，阻塞队列提供了另外4个很是有用的方法：

　　put(E e)：put方法用来向队尾存入元素，若是队列满，则等待；

　　take()：take方法用来从队首取元素，若是队列为空，则等待；

　　offer(E e,long timeout, TimeUnit unit)：offer方法用来向队尾存入元素，若是队列满，则等待必定的时间，当时间期限达到时，若是尚未插入成功，则返回false；不然返回true；

　　poll(long timeout, TimeUnit unit)：poll方法用来从队首取元素，若是队列空，则等待必定的时间，当时间期限达到时，若是取到，则返回null；不然返回取得的元素；

 

　　阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。在阻塞队列不可用时，这两个附加操做提供的4种处理方式，以下表所示。

　　

• 抛出异常：当队列满时，若是再往队列里插入元素，会抛出IllegalStateException（"Queuefull"）异常。当队列空时，从队列里获取元素会抛出NoSuchElementException异常。
• 返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回true，失败返回false。若是是移除方法，则是从队列里取出一个元素，若是有就返回元素，没有则返回null。
• 一直阻塞：当阻塞队列满时，若是生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，若是消费者线程从队列里take元素，队列会阻塞住消费者线程，直到队列不为空。
• 超时退出：当阻塞队列满时，若是生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，若是超过了指定的时间，生产者线程就会退出。

 

二.七种主要的阻塞队列

　　自从Java 1.5以后，在java.util.concurrent包下提供了若干个阻塞队列，主要有如下几个：

　　1.ArrayBlockingQueue：基于数组实现的一个有界阻塞队列，该队列内部维持着一个定长的数据缓冲队列（该队列由数组构成），此队列按照先进先出（FIFO）的原则对元素进行排序，在建立ArrayBlockingQueue对象时必须指定容量大小。ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。

　　而且还能够指定公平性与非公平性，默认状况下为非公平的。所谓公平访问队列是指阻塞的线程，能够按照阻塞的前后顺序访问队列，即先阻塞线程先访问队列。非公平性是对先等待的线程是非公平的，当队列可用时，阻塞的线程均可以争夺访问队列的资格，有可能先阻塞的线程最后才访问队列。为了保证公平性，一般会下降吞吐量。咱们可使用如下代码建立一个公平的阻塞队列。

 

ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);
public ArrayBlockingQueue(int capacity, boolean fair) { 
　　if (capacity <= 0) throw new IllegalArgumentException(); 
　　this.items = new Object[capacity]; 
　　lock = new ReentrantLock(fair); //能够看出访问者的公平性是使用可重入锁实现的
　　notEmpty = lock.newCondition(); 
　　notFull = lock.newCondition();
 }

　　2.LinkedBlockingQueue：基于链表实现的一个有界阻塞队列，内部维持着一个数据缓冲队列（该队列由链表构成），此队列按照先进先出的原则对元素进行排序。当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者当即返回；只有当队列缓冲区达到最大值缓存容量时（能够经过LinkedBlockingQueue的构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程将会被唤醒，反之对于消费者这端的处理也基于一样的原理。在建立LinkedBlockingQueue对象时若是不指定容量大小，则默认大小为Integer.MAX_VALUE。这样的话，若是生产者的速度一旦大于消费者的速度，也许尚未等到队列满阻塞产生，系统内存就有可能已经被消耗殆尽了。

　　LinkedBlockingQueue之因此可以高效的处理并发数据，是由于其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的状况下生产者和消费者能够并行地操做队列中的数据，以此来提升整个队列的并发性能。

　　3.PriorityBlockingQueue：支持优先级排序的无界阻塞队列，以上2种队列都是先进先出队列，而PriorityBlockingQueue却不是，它会按照元素的优先级对元素进行排序，默认状况下元素采起天然顺序排列，也能够经过构造函数传入的Compator对象来决定。而且也是按照优先级顺序出队，每次出队的元素都是优先级最高的元素。在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁。须要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只是在没有可消费的数据时阻塞数据的消费者，所以使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，不然时间一长，会最终耗尽全部的可用堆内存空间。注意，此阻塞队列为无界阻塞队列，即容量没有上限（经过源码就能够知道，它没有容器满的信号标志）。

　　4.DelayQueue：基于PriorityQueue，一种支持延时的获取元素的无界阻塞队列，DelayQueue中的元素只有当其指定的延迟时间到了，才可以从队列中获取到该元素。DelayQueue也是一个无界队列，所以往队列中插入数据的操做（生产者）永远不会被阻塞，而只有获取数据的操做（消费者）才会被阻塞。

　 　5.SynchronousQueue：一个不存储元素的阻塞队列。每个put操做必须等待一个take操做，不然不能继续添加元素。 能够认为SynchronousQueue是一个缓存值为1的阻塞队列，可是SynchronousQueue内部并无数据缓存空间，数据是在配对的生产者和消费者线程之间直接传递的。能够这样来理解：SynchronousQueue是一个传球手，SynchronousQueue不存储数据元素，队列头元素是第一个排队要插入数据的线程，而不是要交换的数据，SynchronousQueue负责把生产者线程处理的数据直接传递给消费者线程，生产者和消费者互相等待对方，握手，而后一块儿离开。它支持公平访问队列。默认状况下线程采用非公平性策略访问队列。在建立公平性访问的SynchronousQueue，若是设置为true，则等待的线程会采用先进先出的顺序访问队列。

　　 6.LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。相对于其余阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。

　　transfer()方法：若是当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法），transfer()方法能够把生产者传入的元素马上传输给消费者；若是没有消费者在等待接收元素，transfer()方法会将元素存放到队列的tail节点，并等到该元素被消费者消费了才返回。

　　transfer()方法的关键代码以下：

Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

　　第一行代码是试图把存放当前元素的s节点做为tail节点，第二行代码是让CPU自旋等待消费者消费元素。由于自旋会消耗CPU，因此自旋必定的次数后使用Thread.yield()方法来暂停当前正在执行的线程，并执行其余线程。

　　tryTransfer()方法：该方法是用来试探生产者传入的元素是否能直接传给消费者，若是没有消费者等待接收元素，则返回false。与transfer()方法的区别：tryTransfer()方法是当即返回（不管消费者是否接收），transfer()方法是必须等到消费者消费了才返回。对于带有时间限制的tryTransfer(E e, long timeout, TimeUnit unit)方法，则是试图把生产者传入的元素直接传给消费者，可是若是没有消费者消费该元素则等待指定的时间以后再返回，若是超时还没消费元素，则返回false，若是在超时时间内消费了元素，则返回true。

　　 7.LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是能够从队列的两端插入和移出元素。双向队列由于多了一个操做队列的入口，在多线程同时入队时，也就减小了一半的竞争。相比其余的阻塞队列，LinkedBlockingDeque多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法，以First单词结尾的方法，表示插入、获取（peek）或移除双端队列的第一个元素。以Last单词结尾的方法，表示插入、获取或移除双端队列的最后一个元素。另外，插入方法add等同于addLast，移除方法remove等效于removeFirst。

 

三.阻塞队列的实现原理

　　 本文以ArrayBlockingQueue为例，其余阻塞队列实现原理可能和ArrayBlockingQueue有一些差异，可是大致思路应该相似，有兴趣的朋友可自行查看其余阻塞队列的实现源码。

　　首先看一下ArrayBlockingQueue类中的几个成员变量：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    /**
     * Serialization ID. This class relies on default serialization
     * even for the items array, which is default-serialized, even if
     * it is empty. Otherwise it could not be declared final, which is
     * necessary here.
     */
    private static final long serialVersionUID = -817911632652898426L;
    /** The queued items */
    final Object[] items;
    /** items index for next take, poll, peek or remove */
    int takeIndex;
    /** items index for next put, offer, or add */
    int putIndex;
    /** Number of elements in the queue */
    int count;
    /** Main lock guarding all access */
    final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    /** Condition for waiting puts */
    private final Condition notFull;
    transient Itrs itrs = null;

 

　　能够看出，ArrayBlockingQueue中用来存储元素的其实是一个数组，takeIndex和putIndex分别表示队首元素和队尾元素的下标，count表示队列中元素的个数。 lock是一个可重入锁，notEmpty和notFull是等待条件。

　　从上述代码中咱们可知，若是队列是空的，消费者会一直等待，当生产者添加元素时，生产者是使用Condition线程间通讯的方法来通知另外一个消费者线程的。当生产者往满的队列里添加元素时会阻塞住生产者，当消费者消费了一个队列中的元素后，会通知生产者当前队列可用。那具体是怎么通知的的？咱们可详细看分析下下面几个方法。

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();//操做以前先上锁
        try {
            while (count == items.length)//当队列满了
                notFull.await();   //则生产者不继续添加，而是将本身阻塞，直到有消费者来消费并将本身唤醒后，才能够继续执行
            enqueue(e);
        } finally {
            lock.unlock();  //释放锁
        }
    }
private void enqueue(E x) {//至关于add()方法
    final Object[] items = this.items;
    items[putIndex] = x;//在队尾添加元素
    if (++putIndex == items.length)//索引自增，若是已经是最后一个位置，从新设置 putIndex = 0
　　　　 putIndex = 0;
    count++;
    notEmpty.signal();
}
public E take() throws InterruptedException {//因为此时并发容器已满，因此生产者生产失败，释放了锁，轮到消费者执行
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly(); //操做前先上锁
        try {
            while (count == 0)//判断容器不为空
                notEmpty.await();
            return dequeue();//调用该方法
        } finally {
            lock.unlock();
        }
    }
private E dequeue() {//至关于remove()
    final Object[] items = this.items;//获取数组容器
    E x = (E) items[takeIndex];//获取队首元素，由于ArrayBlockingQueue是先进先出队列
    items[takeIndex] = null;//将该位置置空
    if (++takeIndex == items.length)//索引自增，若是已经是最后一个位置，从新设置 putIndex = 0
　　　　 takeIndex = 0;
    count--;//将容器中元素个数减一
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();//唤醒其余被阻塞的线程，因为刚才生产者因容器已满而被阻塞掉，这时候就会被该线程唤醒了，唤醒以后就可继续它的生产工做。
    return x;
}

 

 

　　 从put方法的实现能够看出，它先获取了锁，而且获取的是可中断锁，而后判断当前元素个数是否等于数组的长度，若是相等，表示 队列元素已满， 则调用notFull.await()进行等待， 那么当前线程将会被notFull条件对象挂起加到等待队列中，直到队列有空位才会唤醒执行添加操做。但若是队列没有满，那么就直接调用enqueue(e)方法将元素加入到数组队列中。调用tack()方法也是一样的原理。

 

 

 

，所 谓 公平 访问队 列是指阻塞的 线 程，能够按照

阻塞的前后 顺 序 访问队 列，即先阻塞 线 程先 访问队 列。非公平性是 对 先等待的 线 程是非公平

的，当 队 列可用 时 ，阻塞的 线 程均可以争 夺访问队 列的 资 格，有可能先阻塞的 线 程最后才 访问

队 列。 为 了保 证 公平性，一般会下降吞吐量。我 们 可使用如下代 码创 建一个公平的阻塞 队

列。