Java并发编程：阻塞队列

　　三.阻塞队列的实现原理
　　前面谈到了非阻塞队列和阻塞队列中常用的方法，下面来探讨阻塞队列的实现原理，本文以ArrayBlockingQueue为例，其他阻塞队列实现原理可能和ArrayBlockingQueue有一些差别，但是大体思路应该类似，有兴趣的朋友可自行查看其他阻塞队列的实现源码。
　　首先看一下ArrayBlockingQueue类中的几个成员变量：
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>， java.io.Serializable {
private static final long serialVersionUID = -817911632652898426L;
/** The queued items  */
private final E[] items;
/** items index for next take， poll or remove */
private int takeIndex;
/** items index for next put， offer， or add. */
private int putIndex;
/** Number of items in the queue */
private int count;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
private final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
}
　　可以看出，ArrayBlockingQueue中用来存储元素的实际上是一个数组，takeIndex和putIndex分别表示队首元素和队尾元素的下标，count表示队列中元素的个数。
　　lock是一个可重入锁，notEmpty和notFull是等待条件。
　　下面看一下ArrayBlockingQueue的构造器，构造器有三个重载版本：
public ArrayBlockingQueue(int capacity) {
}
public ArrayBlockingQueue(int capacity， boolean fair) {
}
public ArrayBlockingQueue(int capacity， boolean fair，
Collection<? extends E> c) {
}
　　第一个构造器只有一个参数用来指定容量，第二个构造器可以指定容量和公平性，第三个构造器可以指定容量、公平性以及用另外一个集合进行初始化。
　　然后看它的两个关键方法的实现：put()和take()：
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
final E[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == items.length)
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); // propagate to non-interrupted thread
throw ie;
}
insert(e);
} finally {
lock.unlock();
}
}
　　从put方法的实现可以看出，它先获取了锁，并且获取的是可中断锁，然后判断当前元素个数是否等于数组的长度，如果相等，则调用notFull.await()进行等待，如果捕获到中断异常，则唤醒线程并抛出异常。
　　当被其他线程唤醒时，通过insert(e)方法插入元素，后解锁。
　　我们看一下insert方法的实现：
private void insert(E x) {
items[putIndex] = x;
putIndex = inc(putIndex);
++count;
notEmpty.signal();
}
　　它是一个private方法，插入成功后，通过notEmpty唤醒正在等待取元素的线程。
　　下面是take()方法的实现：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == 0)
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to non-interrupted thread
throw ie;
}
E x = extract();
return x;
} finally {
lock.unlock();
}
}
　　跟put方法实现很类似，只不过put方法等待的是notFull信号，而take方法等待的是notEmpty信号。在take方法中，如果可以取元素，则通过extract方法取得元素，下面是extract方法的实现：
　　private E extract() {
　　final E[] items = this.items;
　　E x = items[takeIndex];
　　items[takeIndex] = null;
　　takeIndex = inc(takeIndex);
　　--count;
　　notFull.signal();
　　return x;
　　}
　　跟insert方法也很类似。
　　其实从这里大家应该明白了阻塞队列的实现原理，事实它和我们用Object.wait()、Object.notify()和非阻塞队列实现生产者-消费者的思路类似，只不过它把这些工作一起集成到了阻塞队列中实现。