Java中的阻塞队列

什么是阻塞队列？

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作都支持阻塞的插入和移除方法。 1. 支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不满 2. 支持阻塞的移除方法：意思是队列为空时，获取元素的线程会等待队列变为非空。

插入和移除操作的四种处理方式

方法/处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

Java中的阻塞队列

• ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列

• LinkedBlockingQueue：一个有由链表结构组成的无界阻塞队列

• PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列

• DelayQueue：一个使用优先级队列实现的无界阻塞队列

• SynchronousQueue：一个不存储元素的阻塞队列

• LinkedTransferQueue：一个由链表结构组成的无界阻塞队列

• LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列

ArrayBlockingQueue

是一个用数组实现的有界阻塞队列，此队列按照先进先出的原则对元素进行排序

默认情况下不保证线程公平的访问队列，所谓公平访问队列是指阻塞的线程，可以按照阻塞的先后顺序访问队列，即先阻塞的线程先访问队列。为了公平性往往会降低吞吐量

访问的公平性是使用可重入锁实现的

/**
 * Creates an {@code ArrayBlockingQueue} with the given (fixed)
 * capacity and the specified access policy.
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 *        on insertion or removal, are processed in FIFO order;
 *        if {@code false} the access order is unspecified.
 * @throws IllegalArgumentException if {@code capacity < 1}
 */
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();
}

LinkedBlockingQueue

是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

/**
 * Creates a {@code LinkedBlockingQueue} with a capacity of
 * {@link Integer#MAX_VALUE}.
 */
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

PriorityBlockingQueue

是一个支持优先级的无界阻塞队列(无界只是没有指明界限，最大长度都为Integer.MAX_VALUE，具体可看size返回值，否则会出现问题)，默认情况下元素采用自然顺序升序排列。也是自定义类实现compareTo()方法来指定元素排序规则，或者初始化队列指定构造参数比较器来对元素进行排序

DelayQueue

DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素

应用场景

• 缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了

• 定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，比如TimerQueue就是使用DelayQueue实现的

SynchronousQueue

是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素

它支持公平访问队列。默认情况下采用非公平策略访问队列。

/**
 * Creates a {@code SynchronousQueue} with the specified fairness policy.
 *
 * @param fair if true, waiting threads contend in FIFO order for
 *        access; otherwise the order is unspecified.
 */
public SynchronousQueue(boolean fair) {
    transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}

SynchronousQueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合传递性场景。吞吐量高于LinkedBlockingQueue和ArrayBlockingQueue

LinkedTransferQueue

是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列，多了tryTransfer和transfer方法

transfer方法

如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法），transfer方法可以把生产者传入的元素立刻transfer给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。

tryTransfer方法

用来试探生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回，而transfer方法是必须等待消费者消费了才返回

对于带有时间限制的tryTransfer(e,timeout,unit)方法，试图把生产者传入的元素直接传给消费者，但是如果没有消费者消费该元素则等待指定的时间再返回，如果超时还没消费元素，则返回false，如果在超时时间内消费了元素，则返回true

LinkedBlockingDeque

是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移除元素，双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。

双向队列可以运用在“工作窃取”模式中

阻塞队列的实现原理

使用通知模式实现 ，所谓通知模式，就是当生产者往满的队列里添加元素时会阻塞住生产者，当消费者消费了一个队列中的元素后，会通知生产者当前队列可用。

当往队列里插入一个元素时，如果队列不可用，那么阻塞生产者主要通过LockSupport.park(this)来实现，只有以下四种情况下的一种发生时，该方法才返回

• 与park对应的unpark执行或已经执行时。“已经执行”是指unpark先执行，然后再执行park的情况

• 线程中断时

• 等待完time参数指定的毫秒数时

• 异常现象发生时，这个异常现象没有任何原因

Linux下底层用的是系统方法pthread_cond_wait实现