1. ArrayBlockingQueue简介

在多线程编程过程中，为了业务解耦和架构设计，经常会使用并发容器用于存储多线程间的共享数据，这样不仅可以保证线程安全，还可以简化各个线程操作。例如在“生产者-消费者”问题中，会使用阻塞队列（BlockingQueue）作为数据容器，关于BlockingQueue可以看这篇文章。为了加深对阻塞队列的理解，唯一的方式是对其实验原理进行理解，这篇文章就主要来看看ArrayBlockingQueue和LinkedBlockingQueue的实现原理。

2. ArrayBlockingQueue实现原理

阻塞队列最核心的功能是，能够可阻塞式的插入和删除队列元素。当前队列为空时，会阻塞消费数据的线程，直至队列非空时，通知被阻塞的线程；当队列满时，会阻塞插入数据的线程，直至队列未满时，通知插入数据的线程（生产者线程）。那么，多线程中消息通知机制最常用的是lock的condition机制，关于condition可以看这篇文章的详细介绍。那么ArrayBlockingQueue的实现是不是也会采用Condition的通知机制呢？下面来看看。

2.1 ArrayBlockingQueue的主要属性

ArrayBlockingQueue 的主要属性如下:

  1. /** The queued items */
  2. final Object[] items;
  3. /** items index for next take, poll, peek or remove */
  4. int takeIndex;
  5. /** items index for next put, offer, or add */
  6. int putIndex;
  7. /** Number of elements in the queue */
  8. int count;
  9. /*
  10.  * Concurrency control uses the classic two-condition algorithm
  11.  * found in any textbook.
  12.  */
  13. /** Main lock guarding all access */
  14. final ReentrantLock lock;
  15. /** Condition for waiting takes */
  16. private final Condition notEmpty;
  17. /** Condition for waiting puts */
  18. private final Condition notFull;

从源码中可以看出 ArrayBlockingQueue 内部是采用数组进行数据存储的（属性items），为了保证线程安全，采用的是ReentrantLock lock，为了保证可阻塞式的插入删除数据利用的是 Condition，当获取数据的消费者线程被阻塞时会将该线程放置到 notEmpty 等待队列中，当插入数据的生产者线程被阻塞时，会将该线程放置到 notFull 等待队列中。而 notEmpty 和 notFull 等中要属性在构造方法中进行创建：

  1. public ArrayBlockingQueue(int capacity, boolean fair) {
  2.     if (capacity <= 0)
  3.         throw new IllegalArgumentException();
  4.     this.items = new Object[capacity];
  5.     lock = new ReentrantLock(fair);
  6.     notEmpty = lock.newCondition();
  7.     notFull =  lock.newCondition();
  8. }

接下来，主要看看可阻塞式的put和take方法是怎样实现的。

2.2 put方法详解

put(E e)方法源码如下：

  1. public void put(E e) throws InterruptedException {
  2.     checkNotNull(e);
  3.     final ReentrantLock lock = this.lock;
  4.     lock.lockInterruptibly();
  5.     try {
  6.         //如果当前队列已满，将线程移入到notFull等待队列中
  7.         while (count == items.length)
  8.             notFull.await();
  9.         //满足插入数据的要求，直接进行入队操作
  10.         enqueue(e);
  11.     } finally {
  12.         lock.unlock();
  13.     }
  14. }

该方法的逻辑很简单，当队列已满时（count == items.length）将线程移入到notFull等待队列中，如果当前满足插入数据的条件，就可以直接调用enqueue(e)插入数据元素。enqueue方法源码为：

  1. private void enqueue(E x) {
  2.     // assert lock.getHoldCount() == 1;
  3.     // assert items[putIndex] == null;
  4.     final Object[] items = this.items;
  5.     //插入数据
  6.     items[putIndex] = x;
  7.     if (++putIndex == items.length)
  8.         putIndex = 0;
  9.     count++;
  10.     //通知消费者线程，当前队列中有数据可供消费
  11.     notEmpty.signal();
  12. }

enqueue方法的逻辑同样也很简单，先完成插入数据，即往数组中添加数据（items[putIndex] = x），然后通知被阻塞的消费者线程，当前队列中有数据可供消费（notEmpty.signal()）。

2.3 take方法详解

take方法源码如下：

  1. public E take() throws InterruptedException {
  2.     final ReentrantLock lock = this.lock;
  3.     lock.lockInterruptibly();
  4.     try {
  5.         //如果队列为空，没有数据，将消费者线程移入等待队列中
  6.         while (count == 0)
  7.             notEmpty.await();
  8.         //获取数据
  9.         return dequeue();
  10.     } finally {
  11.         lock.unlock();
  12.     }
  13. }

take 方法也主要做了两步：1. 如果当前队列为空的话，则将获取数据的消费者线程移入到等待队列中；2. 若队列不为空则获取数据，即完成出队操作dequeue。dequeue方法源码为：

  1. private E dequeue() {
  2.     // assert lock.getHoldCount() == 1;
  3.     // assert items[takeIndex] != null;
  4.     final Object[] items = this.items;
  5.     @SuppressWarnings("unchecked")
  6.     //获取数据
  7.     E x = (E) items[takeIndex];
  8.     items[takeIndex] = null;
  9.     if (++takeIndex == items.length)
  10.         takeIndex = 0;
  11.     count--;
  12.     if (itrs != null)
  13.         itrs.elementDequeued();
  14.     //通知被阻塞的生产者线程
  15.     notFull.signal();
  16.     return x;
  17. }

dequeue方法也主要做了两件事情：1. 获取队列中的数据，即获取数组中的数据元素（(E) items[takeIndex]）；2. 通知notFull等待队列中的线程，使其由等待队列移入到同步队列中，使其能够有机会获得lock，并执行完成功退出。

从以上分析，可以看出put和take方法主要是通过 condition 的通知机制来完成可阻塞式的插入数据和获取数据。在理解 ArrayBlockingQueue 后再去理解 LinkedBlockingQueue 就很容易了。