Java中的阻塞队列
1. 什么是阻塞队列
阻塞队列是支持两个附加操作的队列。这两个附加操作就是阻塞式的插入和移除方法。
- 支持阻塞的插入方法：当队列满时，队列会阻塞插入元素的线程，直到队列不满；
- 支持阻塞的移除方法：当队列为空时，队列会阻塞获取元素的线程，直到队列非空。
在阻塞队列不可用时，这两个附加操作提供了4种处理方式：
| 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 | |
|---|---|---|---|---|
| 插入 | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| 移除 | remove() | poll() | take() | poll(time, unit) |
| 检查 | element() | peek() | 不可用 | 不可用 |
下面，解释一下这四种情况具体如何处理：
- 抛出异常：队列满时，再次插入会抛出IllegalStateException（Queue full）异常。队列空时，再次获取元素会抛出NoSuchElementException异常；
- 返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回true。如果是移除方法，则是从队列里取出一个元素，如果没有返回null；
- 一直阻塞：队列满时，如果生产者线程继续put元素，队列就会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列为空时，如果消费者线程继续take元素，那么队列会阻塞消费者线程直到队列不为空；
- 超时退出：阻塞队列满时，如果生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，如果超出了这个时间，生产者线程就会退出。
2. Java里面的阻塞队列
JDK7提供了7个阻塞队列。
2.1 ArrayBlockingQueue
一个由数组结构组成的有界阻塞队列。按照FIFO原则对元素排序。默认情况下不保证线程公平的访问队列，即不保证先阻塞的线程先访问队列。可以通过构造器传入参数构建一个公平的阻塞队列。访问者的公平性是通过可重入锁实现的。
2.2 LinkedBlockingQueue
一个由链表实现的有界阻塞队列。默认最大长度Integer.MAX_VALUE。按照FIFO对元素进行排序。
2.3 PriorityBlockingQueue
一个支持优先级的无界阻塞队列，默认情况下采用自然排序升序排列，也可以自定义排序规则。
2.4 DelayQueue
一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue实现。
2.5 SynchronousQueue
不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能添加元素。支持公平访问队列。
2.6 LinkedTransferQueue
由链表结构组成的无界阻塞队列，比其他阻塞队列多了一个tryTransfer和transfer方法。
2.7 LinkedBlockingDeque
链表结构组成的双向阻塞队列。
3. ArrayBlockingQueue实现原理
阻塞队列是通过通知模式实现生产者和消费者之间的通信的，当生产者向一个满的队列put数据的时候会被阻塞，当消费者消费了一个队列元素后，会通知生产者当前队列可用。看一下源码：
构造器：
final Object[] items; // 存放队列元素的数组
private final Condition notEmpty; // 等待take的Condition
private final Condition notFull; // 等待put的Condition
final ReentrantLock lock; // 可重入锁
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity]; // 初始化存放队列的数组
lock = new ReentrantLock(fair); // fair:true 公平锁 fair:false 非公平锁(默认)
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
存储元素
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) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
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() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal(); // 刚取出了一个元素，队列肯定不为空，通知生产者你可以来放入元素了
return x;
}
从源码可以看到生产者放入元素的时候，如果队列已满，则阻塞生产者放入元素，直到有消费者消费了队列元素就通知生产者可以放入了。同理，当消费者取出元素的时候，如果队列为空，则阻塞消费者，直到生产者放入了元素则通知消费者你可以继续取元素了。这是一个典型的通知模式。
参考：
《Java并发编程的艺术》
