Java并发容器之ArrayBlockingQueue

一、ArrayBlockingQueue简介

    ArrayBlockingQueue是一个数组支持的有界阻塞队列。按照FIFO的原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。

    这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。

    此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。默认情况下，不保证是这种排序。然而，通过将公平性 (fairness) 设置为
true 而构造的队列允许按照 FIFO 顺序访问线程。公平性通常会降低吞吐量，但也减少了可变性和避免了“不平衡性”。

二、ArrayBlockingQueue结构

    如图，ArrayBlockingQueue使用Object[]数组 items 来存放队列元素。takeIndex是出队下标，putIndex是入队下标，count统计元素个数，这些个定义并未使用volatile修饰，因为这些访问是使用重入锁ReentrantLock lock锁住再进行操作的。另外，lock生成了两个condition，分别对非空和非满情况进行唤醒操作。
    构造方法如下：

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);
notEmpty = lock.newCondition();
notFull =  lock.newCondition();
}

    可见，默认是非公平锁，这样效率更高一些。同时，构造了capacity大小的数组，这个数组是final的，因此队列一旦构造，其容量就不可改变。同时初始化了非空Condition和非满Condition。

三、offer操作

插入元素，失败返回false，否则true

public boolean offer(E e) {
checkNotNull(e);			//e为null 则抛出 NullPointerException 异常
final ReentrantLock lock = this.lock;   //获取重入锁
lock.lock();				//加锁
try {
if (count == items.length)      	//如果队列已经满了 返回false
return false;
else {
enqueue(e);			//否则插入元素
return true;
}
} finally {
lock.unlock();			//解锁
}
}

private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0; //注意，下一个插入index为0，因为这里使用了循环数组
count++;
notEmpty.signal(); //有了元素插入，非空条件激活，唤醒被take()阻塞的线程
}

    因为加了锁，所以不存在竞争条件。另外这个队列是使用循环数组实现的，所以计算下一个元素储存位置下标的时候有点特殊。另外，最后调用了notEmpty.signal()方法，激活调用了notEmpty.wait()而阻塞后放入notEmpty阻塞队列中的线程。

四、put操作

    在队列尾部添加元素，如果队列满则阻塞等待有空位置插入，然后返回。

public void put(E e) throws InterruptedException {
checkNotNull(e);			//检查非空
final ReentrantLock lock = this.lock;	//获取锁
lock.lockInterruptibly();		//如果线程未被中断，则获取锁
try {
while (count == items.length)	//如果线程满了，就让线程进入notFull的阻塞队列
notFull.await();
enqueue(e);				//线程被唤醒，获取锁，插入元素
} finally {
lock.unlock();			//解锁
}
}

    关键点是，如果队列满了，线程会阻塞，知道空了时候，notFull唤醒线程。

    另一个问题是为什么需要使用lock.lockInterruptibly()方法而不是Lock方法。这里我们查看condition.await()方法的详细信息，里面有这么一条：如果当前线程1）在进入此方法时已经设置了该线程的中断状态，或者2）在支持等待和中断线程挂起时，线程被中断；则抛出 InterruptedException ，并清除当前线程的中断状态。因此，与其在获得锁之后发现中断而抛出异常退出，还不如在加锁的时候就先看中断标志是不是被设置了，如果是，则清除中断状态、抛出异常退出。

五、poll操作

    从队头获取并移除元素，如果队列为空，则返回null。

public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();	//如果有元素，则获取第一个
} finally {
lock.unlock();
}
}

private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];		//获取元素
items[takeIndex] = null;		//清空该index中元素
if (++takeIndex == items.length)	//循环数组
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();			//激活需要放入元素的线程
return x;
}

    poll操作也会加锁，在取出元素后，队列非空，因此会唤醒因为队列满了而阻塞的插入线程。

六、take操作

    从队头获取元素，如果队列为空则阻塞。

public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;	//获取锁对象
lock.lockInterruptibly();		//如果未被中断，则获取锁，否则抛出异常，清除中断位，退出
try {
while (count == 0)			//如果队列为空，则阻塞等待唤醒
notEmpty.await();
return dequeue();			//唤醒后 且队列不为空，则出队
} finally {
lock.unlock();			//解锁
}
}

七、peek操作

public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}

    没啥好说的哈哈哈

八、size操作

public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}

    一目了然。

九、总结

    ArrayBlockingQueue通过重入锁和两个条件condition，实现了同时只有一个线程入队或出队，有点Synchronized的味道。同时，take和put操作，实现了条件阻塞，当队列空或满时阻塞而不是退出。

    因为每次计算前都是加锁的，所以ArrayBlockingQueue是精确的。