ArrayBlockingQueue 源码阅读
ArrayBlockingQueue 简介
阻塞队列就是典型的生产者-消费者模型，它可以做到以下几点:
总结一下：阻塞队列就说基于非空和非满两个条件实现生产者和消费者之间的交互，尽管这些交互流程和等待通知的机制实现非常复杂，好在 Doug Lea 的操刀之下已将阻塞队列的细节屏蔽，我们只需调用 put、take、offfer、poll 等 API 即可实现多线程之间的生产和消费。
这也使得阻塞队列在多线程开发中有着广泛的运用，最常见的例子无非是我们的线程池,从源码中我们就能看出当核心线程无法及时处理任务时，这些任务都会扔到 workQueue 中。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {// …}
ArrayBlockingQueue 源码分析
整体设计
在了解 ArrayBlockingQueue 的具体细节之前，我们先来看看 ArrayBlockingQueue 的类图。
从图中我们可以看出，ArrayBlockingQueue 继承了阻塞队列 BlockingQueue 这个接口，不难猜出通过继承 BlockingQueue 这个接口之后，ArrayBlockingQueue 就拥有了阻塞队列那些常见的操作行为。
同时， ArrayBlockingQueue 还继承了 AbstractQueue 这个抽象类，这个继承了 AbstractCollection 和 Queue 的抽象类，从抽象类的特定和语义我们也可以猜出，这个继承关系使得 ArrayBlockingQueue 拥有了队列的常见操作。
所以我们是否可以得出这样一个结论，通过继承 AbstractQueue 获得队列所有的操作模板，其实现的入队和出队操作的整体框架。然后 ArrayBlockingQueue 通过继承 BlockingQueue 获取到阻塞队列的常见操作并将这些操作实现，填充到 AbstractQueue 模板方法的细节中，由此 ArrayBlockingQueue 成为一个完整的阻塞队列。
为了印证这一点，我们到源码中一探究竟。首先我们先来看看 AbstractQueue，从类的继承关系我们可以大致得出，它通过 AbstractCollection 获得了集合的常见操作方法，然后通过 Queue 接口获得了队列的特性。
public abstract class AbstractQueue<E>
extends AbstractCollection<E>
implements Queue<E> {
//…
}
对于集合的操作无非是增删改查，所以我们不妨从添加方法入手，从源码中我们可以看到，它实现了 AbstractCollection 的 add 方法，其内部逻辑如下:
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
而 AbstractQueue 中并没有对 Queue 的 offer 的实现，很明显这样做的目的是定义好了 add 的核心逻辑，将 offer 的细节交由其子类即我们的 ArrayBlockingQueue 实现。
到此，我们对于抽象类 AbstractQueue 的分析就结束了，我们继续看看 ArrayBlockingQueue 中另一个重要的继承接口 BlockingQueue。
点开 BlockingQueue 之后，我们可以看到这个接口同样继承了 Queue 接口，这就意味着它也具备了队列所拥有的所有行为。同时，它还定义了自己所需要实现的方法。
public interface BlockingQueue<E> extends Queue<E> {
//元素入队成功返回true，反之则会抛出异常IllegalStateException
boolean add(E e);
//元素入队成功返回true，反之返回false
boolean offer(E e);
//元素入队成功则直接返回，如果队列已满元素不可入队则将线程阻塞，因为阻塞期间可能会被打断，所以这里方法签名抛出了InterruptedException
void put(E e) throws InterruptedException;
//和上一个方法一样,只不过队列满时只会阻塞单位为unit，时间为timeout的时长，如果在等待时长内没有入队成功则直接返回false。
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
//从队头取出一个元素，如果队列为空则阻塞等待，因为会阻塞线程的缘故，所以该方法可能会被打断，所以签名定义了InterruptedException
E take() throws InterruptedException;
//取出队头的元素并返回，如果当前队列为空则阻塞等待timeout且单位为unit的时长，如果这个时间段没有元素则直接返回null。
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
//获取队列剩余元素个数
int remainingCapacity();
//删除我们指定的对象，如果成功返回true，反之返回false。
boolean remove(Object o);
//判断队列中是否包含指定元素
public boolean contains(Object o);
//将队列中的元素全部存到指定的集合中
int drainTo(Collection<? super E> c);
//转移maxElements个元素到集合中
int drainTo(Collection<? super E> c, int maxElements);
}
了解了 BlockingQueue 的常见操作后，我们就知道了 ArrayBlockingQueue 通过继承 BlockingQueue 的方法并实现后，填充到 AbstractQueue 的方法上，由此我们便知道了上文中 AbstractQueue 的 add 方法的 offer 方法是哪里是实现的了。
public boolean add(E e) {
//AbstractQueue的offer来自下层的ArrayBlockingQueue从BlockingQueue继承并实现的offer方法
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
初始化
了解 ArrayBlockingQueue 的细节前，我们不妨先看看其构造函数，了解一下其初始化过程。从源码中我们可以看出 ArrayBlockingQueue 有 3 个构造方法，而最核心的构造方法就是下方这一个。
// capacity 表示队列初始容量，fair 表示 锁的公平性
public ArrayBlockingQueue(int capacity, boolean fair) {
//如果设置的队列大小小于0，则直接抛出IllegalArgumentException
if (capacity <= 0)
throw new IllegalArgumentException();
//初始化一个数组用于存放队列的元素
this.items = new Object[capacity];
//创建阻塞队列流程控制的锁
lock = new ReentrantLock(fair);
//用lock锁创建两个条件控制队列生产和消费
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
另外两个构造方法都是基于上述的构造方法，默认情况下，我们会使用下面这个构造方法，该构造方法就意味着 ArrayBlockingQueue 用的是非公平锁，即各个生产者或者消费者线程收到通知后，对于锁的争抢是随机的。
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
还有一个不怎么常用的构造方法，在初始化容量和锁的非公平性之后，它还提供了一个 Collection 参数，从源码中不难看出这个构造方法是将外部传入的集合的元素在初始化时直接存放到阻塞队列中。
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
//初始化容量和锁的公平性
this(capacity, fair);
final ReentrantLock lock = this.lock;
//上锁并将c中的元素存放到ArrayBlockingQueue底层的数组中
lock.lock();
try {
int i = 0;
try {
//遍历并添加元素到数组中
for (E e : c) {
checkNotNull(e);
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
//记录当前队列容量
count = i;
//更新下一次put或者offer或用add方法添加到队列底层数组的位置
putIndex = (i == capacity) ? 0 : i;
} finally {
//完成遍历后释放锁
lock.unlock();
}
}
阻塞式获取和新增元素
ArrayBlockingQueue 阻塞式获取和新增元素对应的就是生产者-消费者模型，虽然它也支持非阻塞式获取和新增元素（例如 poll() 和 offer(E e) 方法），但一般不会使用。
ArrayBlockingQueue 阻塞式获取和新增元素的方法为：
- put(E e)：将元素插入队列中，如果队列已满，则该方法会一直阻塞，直到队列有空间可用或者线程被中断。
- take() ：获取并移除队列头部的元素，如果队列为空，则该方法会一直阻塞，直到队列非空或者线程被中断。
这两个方法实现的关键就是在于两个条件对象 notEmpty(非空) 和 notFull（非满），这个我们在上文的构造方法中有提到。
假设我们的代码消费者先启动，当它发现队列中没有数据，那么非空条件就会将这个线程挂起，即等待条件非空时挂起。然后 CPU 执行权到达生产者，生产者发现队列中可以存放数据，于是将数据存放进去，通知此时条件非空，此时消费者就会被唤醒到队列中使用 take 等方法获取值了。
随后的执行中，生产者生产速度远远大于消费者消费速度，于是生产者将队列塞满后再次尝试将数据存入队列，发现队列已满，于是阻塞队列就将当前线程挂起，等待非满。然后消费者拿着 CPU 执行权进行消费，于是队列可以存放新数据了，发出一个非满的通知，此时挂起的生产者就会等待 CPU 执行权到来时再次尝试将数据存到队列中。
简单了解阻塞队列的基于两个条件的交互流程之后，我们不妨看看 put 和 take 方法的源码。
public void put(E e) throws InterruptedException {
//确保插入的元素不为null
checkNotNull(e);
//加锁
final ReentrantLock lock = this.lock;
//这里使用lockInterruptibly()方法而不是lock()方法是为了能够响应中断操作，如果在等待获取锁的过程中被打断则该方法会抛出InterruptedException异常。
lock.lockInterruptibly();
try {
//如果count等数组长度则说明队列已满，当前线程将被挂起放到AQS队列中，等待队列非满时插入（非满条件）。
//在等待期间，锁会被释放，其他线程可以继续对队列进行操作。
while (count == items.length)
notFull.await();
//如果队列可以存放元素，则调用enqueue将元素入队
enqueue(e);
} finally {
//释放锁
lock.unlock();
}
}
put 方法内部调用了 enqueue 方法来实现元素入队，我们继续深入查看一下 enqueue 方法的实现细节：
private void enqueue(E x) {
//获取队列底层的数组
final Object[] items = this.items;
//将putindex位置的值设置为我们传入的x
items[putIndex] = x;
//更新putindex，如果putindex等于数组长度，则更新为0
if (++putIndex == items.length)
putIndex = 0;
//队列长度+1
count++;
//通知队列非空，那些因为获取元素而阻塞的线程可以继续工作了
notEmpty.signal();
}
从源码中可以看到入队操作的逻辑就是在数组中追加一个新元素，整体执行步骤为:
自此我们了解了 put 方法的流程，为了更加完整的了解 ArrayBlockingQueue 关于生产者-消费者模型的设计，我们继续看看阻塞获取队列元素的 take 方法。
public E take() throws InterruptedException {
//获取锁
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//如果队列中元素个数为0，则将当前线程打断并存入AQS队列中，等待队列非空时获取并移除元素（非空条件）
while (count == 0)
notEmpty.await();
//如果队列不为空则调用dequeue获取元素
return dequeue();
} finally {
//释放锁
lock.unlock();
}
}
理解了 put 方法再看 take 方法就很简单了，其核心逻辑和put 方法正好是相反的，比如 put 方法在队列满的时候等待队列非满时插入元素（非满条件），而take 方法等待队列非空时获取并移除元素（非空条件）。
take 方法内部调用了 dequeue 方法来实现元素出队，其核心逻辑和 enqueue 方法也是相反的。
private E dequeue() {
//获取阻塞队列底层的数组
final Object[] items = this.items;
@SuppressWarnings("unchecked")
//从队列中获取takeIndex位置的元素
E x = (E) items[takeIndex];
//将takeIndex置空
items[takeIndex] = null;
//takeIndex向后挪动，如果等于数组长度则更新为0
if (++takeIndex == items.length)
takeIndex = 0;
//队列长度减1
count—;
if (itrs != null)
itrs.elementDequeued();
//通知那些被打断的线程当前队列状态非满，可以继续存放元素
notFull.signal();
return x;
}
非阻塞式获取和新增元素
ArrayBlockingQueue 非阻塞式获取和新增元素的方法为：
- offer(E e)：将元素插入队列尾部。如果队列已满，则该方法会直接返回 false，不会等待并阻塞线程。
- poll()：获取并移除队列头部的元素，如果队列为空，则该方法会直接返回 null，不会等待并阻塞线程。
- add(E e)：将元素插入队列尾部。如果队列已满则会抛出 IllegalStateException 异常，底层基于 offer(E e) 方法。
- remove()：移除队列头部的元素，如果队列为空则会抛出 NoSuchElementException 异常，底层基于 poll()。
- peek()：获取但不移除队列头部的元素，如果队列为空，则该方法会直接返回 null，不会等待并阻塞线程。
先来看看 offer 方法，逻辑和 put 差不多，唯一的区别就是入队失败时不会阻塞当前线程，而是直接返回 false。
public boolean offer(E e) {
//确保插入的元素不为null
checkNotNull(e);
//获取锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//队列已满直接返回false
if (count == items.length)
return false;
else {
//反之将元素入队并直接返回true
enqueue(e);
return true;
}
} finally {
//释放锁
lock.unlock();
}
}
poll 方法同理，获取元素失败也是直接返回空，并不会阻塞获取元素的线程。
public E poll() {
final ReentrantLock lock = this.lock;
//上锁
lock.lock();
try {
//如果队列为空直接返回null，反之出队返回元素值
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
add 方法其实就是对于 offer 做了一层封装，如下代码所示，可以看到 add 会调用没有规定时间的 offer，如果入队失败则直接抛异常。
public boolean add(E e) {
//调用下方的add
return super.add(e);
}
public boolean add(E e) {
//调用offer如果失败直接抛出异常
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
remove 方法同理，调用 poll，如果返回 null 则说明队列没有元素，直接抛出异常。
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
peek() 方法的逻辑也很简单，内部调用了 itemAt 方法。
public E peek() {
//加锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//当队列为空时返回 null
return itemAt(takeIndex);
} finally {
//释放锁
lock.unlock();
}
}
//返回队列中指定位置的元素
@SuppressWarnings("unchecked")
final E itemAt(int i) {
return (E) items[i];
}
指定超时时间内阻塞式获取和新增元素
在 offer(E e) 和 poll() 非阻塞获取和新增元素的基础上，设计者提供了带有等待时间的 offer(E e, long timeout, TimeUnit unit) 和 poll(long timeout, TimeUnit unit)，用于在指定的超时时间内阻塞式地添加和获取元素。
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
checkNotNull(e);
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//队列已满，进入循环
while (count == items.length) {
//时间到了队列还是满的，则直接返回false
if (nanos <= 0)
return false;
//阻塞nanos时间，等待非满
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
可以看到，带有超时时间的 offer 方法在队列已满的情况下，会等待用户所传的时间段，如果规定时间内还不能存放元素则直接返回 false。
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//队列为空，循环等待，若时间到还是空的，则直接返回null
while (count == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
return dequeue();
} finally {
lock.unlock();
}
}
同理，带有超时时间的 poll 也一样，队列为空则在规定时间内等待，若时间到了还是空的，则直接返回 null。
判断元素是否存在
ArrayBlockingQueue 提供了 contains(Object o) 来判断指定元素是否存在于队列中。
public boolean contains(Object o) {
//若目标元素为空，则直接返回 false
if (o == null) return false;
//获取当前队列的元素数组
final Object[] items = this.items;
//加锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 如果队列非空
if (count > 0) {
final int putIndex = this.putIndex;
//从队列头部开始遍历
int i = takeIndex;
do {
if (o.equals(items[i]))
return true;
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
return false;
} finally {
//释放锁
lock.unlock();
}
}
