java阻塞队列ArrayBlockingQueue源码分析

concurrent包的BlockingQueue接口派生类适用于在多线程环境下实现生产者-消费者模式，下面来看一下底层基于数组的阻塞队列ArrayBlockingQueue。

类结构图如下：



ArrayBlockingQueue使用一个数组存储元素，使用ReentrantLock锁实现多线程的并发访问。并且提供了一下四组操作元素的方法，每组操作均有自己特性：



比如在向队列放元素时，此时队列数组满了，如果调用add方法，会立即抛出一个异常，如果调用offer方法，则会立即返回一个false，而如果调用put方法，则会导致生产者线程阻塞住，直到队列中有空闲位置并且成功放入后为止，如果调用带时间参数的offer方法，则在指定时间内不能添加成功则返回false。

由于ArrayBlockingQueue使用ReentrantLock进行并发控制，因此可以指定采用公平锁还是非公平锁，构造方法如下：

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();
}

注意到构造方法中还有两个Condition，是加在ReentrantLock上的等待通知机制，类似于jvm内置对象锁上的wait和notify方法，具体实现细节在此不做追究，下面来看一下它的用法：

public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
insert(e);
} finally {
lock.unlock();
}
}

在put方法中，如果队列满了，那么调用ReentrantLock锁上notFull这个Condition的await()方法，意思是当前线程释放lock锁，让出CPU时间片，并进入lock锁的waiting队列（不要与咱们将的阻塞队列弄混了）进行等待；等其他线程调用了此lock锁的notFull Condition的signal()方法时，会依照jvm的唤醒算法唤醒一个线程并获得锁，此时该调用await()方法的线程将有机会被唤醒，并继续执行put()方法。

signal()方法会在take()方法中被调用：

public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return extract();
} finally {
lock.unlock();
}
}

private E extract() {
final Object[] items = this.items;
E x = this.<E>cast(items[takeIndex]);
items[takeIndex] = null;
takeIndex = inc(takeIndex);
--count;
notFull.signal();
return x;
}

回过头来再注意一下上文中的put方法，此方法是在接口BlockingQueue中声明的：

//BlockingQueue接口
void put(E e) throws InterruptedException;

可见BlockingQueue对其的规范为应当可以响应中断异常，因此上文中的put方法源码才会使用ReentrantLock的可中断锁 ：

lock.lockInterruptibly()

。

另外，由于ArrayBlockingQueue在每个操作元素的方法中都加了锁，所以在并发环境下，无法做到生产者与生产者、生产者与消费者、消费者与消费者之间真正的并行！