前言

并发编程最佳学习路线
【Java多线程】高并发修炼基础之高并发必须了解的概念
【Java多线程】了解线程的锁池和等待池概念
【Java多线程】了解Java锁机制
【Java多线程】线程通信

【Java基础】多线程从入门到掌握-第十五节.使用Concurrent集合
【Java多线程】JUC之线程池(一)与线程池的初识第四节.线程池的工作队列

---

一.JDK提供的并发容器

二.特点

ArrayBlockingQueue：由数组实现的有界阻塞队列，在初始化时必须指定容器大小，按照FIFO的方式存储元素。内部使用ReentrantLock和Condition实现，支持公平锁和非公平锁。

• 队列的容量一旦在构造时指定就无法更改
• 插入元素时，在队尾进行；删除元素时，在队首进行；
• 队列满时，写入元素会阻塞线程；队列空时，移除元素也会阻塞线程；
• 支持公平/非公平策略，默认为非公平策略。

三.继承关系

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    

四.主要属性

   //内部数组
    final Object[] items;
    // 下一个待删除元素的索引: take, poll, peek, remove方法使用
    int takeIndex;
    //下一个待插入元素的索引: put, offer, add方法使用
    int putIndex;
    //当前队列中元素的个数
    int count;
    //唯一全局可重入独占“”：掌管所有读写操作的锁
    final ReentrantLock lock;
    //两个等待队列
    /** 取元素条件队列：队列为空时,用于阻塞读线程,唤醒写线程 */
    private final Condition notEmpty;
    /** 写元素条件队列：队列已满时,用于阻塞写线程,唤醒读线程 */
    private final Condition notFull
    // Itrs表示队列和迭代器之间的共享数据，其实用来存储多个迭代器实例的
    transient Itrs itrs = null;

• 可以看到，ArrayBlockingQueue 使用可重入锁 ReentrantLock实现的访问公平性，通过2个 Condition保证了写入和获取元素的等待通知

五.环形队列

从上面的入队/出队操作，可以看出，ArrayBlockingQueue的内部数组其实是一种环形结构。

• 假设ArrayBlockingQueue的容量大小为6，我们来看下整个入队过程：

1. 初始时
   

2. 插入元素“9”
   

3. 插入元素“2”、“10”、“25”、“93”
   

4. 插入元素 “90”
   注意，此时再插入一个元素“90”，则putIndex变成6，等于队列容量6，由于是循环队列，所以会将takeIndex重置为0：
   

5. 这时队列已经满了（count==6），如果再有线程尝试插入元素，并不会覆盖原有值，而是被阻塞。

---

我们再来看下出队过程：

1. 出队元素“9”

2. 出队元素“2”、“10”、“25”、“93”
   

3. 出队元素“90”
   注意，此时再出队一个元素“90”，则takeIndex变成6，等于队列容量6，由于是循环队列，所以会将takeIndex重置为0：
   
   这时队列已经空了（count==0），如果再有线程尝试出队元素，则会被阻塞。

六.构造方法

	 // 必须指定初始容量， 默认采用非公平策略
    public ArrayBlockingQueue(int capacity) {
    
        this(capacity, false);//默认构造非公平锁的阻塞队列 
    }

	// 指定队列初始容量和公平/非公平策略的构造器.
    public ArrayBlockingQueue(int capacity, boolean fair) {
    
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
       //初始化ReentrantLock重入锁，出队入队拥有这同一个锁 
        lock = new ReentrantLock(fair);
        //初始化非空等待队列
        notEmpty = lock.newCondition();
        //初始化非满等待队列 
        notFull =  lock.newCondition();
    }
	/**
	 * 根据已有集合构造队列
	 */
    public ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c) {
    
        this(capacity, fair);

        final ReentrantLock lock = this.lock;
        lock.lock(); // Lock only for visibility, not mutual exclusion
        try {
    
            int i = 0;
            try {
    
                for (E e : c) {
      //遍历添加指定集合的元素
                    checkNotNull(e);// 不能有null元素
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
    
            //如果传入集合的个数超过了容量，抛出异常被catch，最多放capacity个
                throw new IllegalArgumentException();
            }
            //循环结束，i刚好是写入元素的个数
            count = i;
            //修改 putIndex 为 c 的容量 +1,如果队列已满，则重置puIndex索引为0
            putIndex = (i == capacity) ? 0 : i; 
        } finally {
    
            lock.unlock();
        }
    }

可以看到，有3种构造函数：

• 默认的构造函数需指定队列容量，默认为非公平锁
• 第2种构造函数中，使用 ReentrantLock 创建了 2 个 Condition 锁
• 第3种构造函数可以在创建队列时，将指定的元素添加到队列中

七.入队

入队核心方法-void enqueue(E x)

核心思想:

1. 将元素x置入数组中。
2. 计算下一个元素应该存放的下标位置。
3. 元素个数器递增，这里count前加了锁，值都是从主内存中获取，不会存在内存不可见问题，并且更新也会直接刷新回主内存中。
4. 最后唤醒在条件队列notEmpty因取出元素(take)而被阻塞的一个线程。

//入队操作
private void enqueue(E x) {
    
    final Object[] items = this.items;
    //通过putIndex索引直接将元素添加到数组items中
    items[putIndex] = x;
    //下一个元素应该存放的下标位置：当putIndex索引大小等于数组长度时，将putIndex重置为0
    //当队列索引（从0开始）与数组长度相等时，下次就需要从数组头部重写开始写入
    if (++putIndex == items.length) putIndex = 0;
    count++; // 元素个数+1
    notEmpty.signal();// 唤醒一个notEmpty上的等待线程(可以来队列取元素了)
}

可以看到，enqueue(E) 方法会将元插入到数组尾部。

• 如果插入元素后队列满了，就重置 putIndex 为 0，添加后调用notEmpty.signal()通知唤醒阻塞在取出元素(take)的线程。

阻塞式入队-void put(E e)

• put()用于向队尾阻塞式插入元素，如果队列未满则插入，如果队列已满，则阻塞当前线程直到队列为空闲，或者元素被其他线程取出。
  • 如果线程在阻塞时被其他线程设置了中断标志，则抛出InterruptedException异常并返回。

	//put操作将向队尾插入元素，如果队列未满则插入，如果队列已满，则阻塞当前线程直到队列不满。
	//如果线程在阻塞时被其他线程设置了中断标志，则抛出InterruptedException异常并返回。
   public void put(E e) throws InterruptedException {
    
     checkNotNull(e);
     final ReentrantLock lock = this.lock; // 唯一锁
     // 可响应中断式地获取锁
     lock.lockInterruptibly();
     try {
    
			  //如果队列已满，则将当前线程包装为等待节点置入notFull的条件队列中。这里必须用while，防止虚假唤醒
            while (count == items.length)
            notFull.await();
           // 队列非满，或者被消费者线程唤醒了，执行入队操作,往队尾写入一个元素,然后唤醒等待在notEmpty条件队列的首节点
            enqueue(e);
      } finally {
    
            lock.unlock();// 解锁
      }
  }

• 可以看到，put() 方法可以响应中断，当队列满了，就调用notFull.await() 阻塞等待，等有消费者获取元素后继续执行；
  • 可以插入元素时还是调用enqueue(E)。

非阻塞入队- boolean add(E e)

public boolean add(E e) {
    
    return super.add(e);
}

//super.add() 的实现
public boolean add(E e) {
    
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

• add(E) 调用了父类的方法，而父类里调用的是offer(E)，如果返回 false 就抛出异常。

非阻塞入队-boolean offer(E e)

• offer(E e)是非阻塞的方法，插入元素到队尾
  • 如果队列未满，则插入成功并返回true， 如果队列已满则返回false。

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

可以看到 offer(E) 方法要先获取锁，如果当前队列中元素已满，就立即返回 false，这点比 add() 友好一些；

• 如果没满就调用 enqueue(E)入队：

阻塞式超时入队-boolean offer(E,long,TimeUnit)

• 在put()阻塞式插入方法的基础上额外增加超时功能，传入一个timeout，获取不到而阻塞的时候，如果时间到了，即使还获取不到，也只能立即返回false。

public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
    

    if (e == null) throw new NullPointerException();
    long nanos = unit.toNanos(timeout);
    //获取可中断锁
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
    
        while (count == items.length) {
    
        	//如果队列已满,但是时间到了,直接返回false
            if (nanos <= 0)
                return false;
              //阻塞当前线程指定纳秒数,并更新剩余时间       
            nanos = notFull.awaitNanos(nanos);
        }
        //队列非满，或者在设定时间内被消费者线程唤醒了,执行入队操作,后唤醒等待在notEmpty条件队列的首节点
        enqueue(e);
        return true;
    } finally {
    
        lock.unlock();//释放锁
    }
}

• 可以看到 offer() 和 put() 方法很相似，不同之处在于需要设置等待超时时间，超时未写入，就返回 false；否则调用enqueue(E)入队，然后返回 true。

八.出队

出队核心方法-E dequeue()

核心思想:

1. 获取元素，并将当前位置置null。
2. 重新设置队头下标。
3. 元素计数器递减。
4. 更新迭代器中的元素数据，itrs默认情况下都是为null的，只有使用迭代器的时候才会实例化Itrs。
5. 唤醒在条件队列notFull因写入操作(put)而被阻塞的一个线程。

 //删除队列头元素并返回
private E dequeue() {
    
  	//拿到当前数组的数据
    final Object[] items = this.items;
     //获取要删除的对象
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    //将数组中takeIndex索引位置设置为null
    items[takeIndex] = null;
    //takeIndex索引加1并判断是否与数组长度相等，如果相等说明队列已空,重置takeIndex为0
    if (++takeIndex == items.length) takeIndex = 0;
    count--;//队列个数减1
    // 更新迭代器中的元素数据，itrs只用在使用迭代器的时候才实例化
    if (itrs != null)
        itrs.elementDequeued();//同时更新迭代器中的元素数据
    notFull.signal(); // 唤醒一个notFull上的等待线程(可以插入元素到队列了)
    return x;
}

默认情况下dequeue()方法会从队首移除元素（即 takeIndex 位置）。

• 移除后会向后移动 takeIndex，如果已经到队尾，就归零（takeIndex =0）。结合前面添加元素时的归零，可以看到，其实 ArrayBlockingQueue 是个环形数组。

• 然后调用itrs. elementDequeued()，这个 itrs 是 ArrayBlockingQueue 的内部类 Itrs 的对象，看起来像是个迭代器，实际上它的作用是保证循环数组迭代时的正确性，具体实现比较复杂，这里暂不介绍。

阻塞式出队-E take()

• take()用于获取当前队列头部元素并移除，如果队列为空则阻塞当前线程直到队列不为空，退出阻塞时返回获取的元素。

	//从队列头部删除，队列没有元素就阻塞，可中断
  public E take() throws InterruptedException {
    
     final ReentrantLock lock = this.lock; 
     // 可响应中断式地获取锁
     lock.lockInterruptibly();
     try {
    
      //如果队列为空，则将当前写线程包装为等待节点加入notEmpty的条件队列中。这里必须用while，防止虚假唤醒
       while (count == 0) 
           notEmpty.await();
           // 队列非空，或者被生产者线程唤醒了，执行出队操作，出队时唤醒notEmpty的条件队列中的首节点
           return dequeue();
      } finally {
    
            lock.unlock();// 释放锁
     }
}

• take() 方法可以响应中断，与 poll() 不同的是，如果队列为空会一直阻塞等待，直到中断或者有元素，有元素时还是调用dequeue()方法入队。

非阻塞式出队-E poll()

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

//poll方法，该方法获取并移除此队列的头元素，若队列为空，则返回 null
    public E poll() {
    
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
    
            // 如果为空，返回null， 否则执行出队操作
            return (count == 0) ? null : dequeue();
        } finally {
    
            lock.unlock();
        }
    }

• poll() 如果在队列中没有元素时会立即返回 null；如果有元素调用 dequeue()出队:

阻塞式超时出队-E poll(timeout, unit)

• 在take()阻塞式获取方法的基础上额外增加超时功能，传入一个timeout，获取元素超时会立即返回null

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
    
        while (count == 0) {
    
             // 队列仍为空，但是时间到了，必须返回了
            if (nanos <= 0)
                return null;
                  // 在条件队列里等着，但是需要更新时间
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally {
    
        lock.unlock();
    }
}

• 带参数的 poll() 方法相当于无参 poll() 和 take() 的中和版，允许阻塞一段时间，如果在阻塞一段时间还没有元素写入队列，就返回 null。

阻塞式出队-E peek()

• 直接获取队首元素，只获取不出队。可能返回为null（队列为空）

public E peek() {
    
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
    
        return itemAt(takeIndex);   //直接返回当前队列的头元素，但不删除
    } finally {
    
        lock.unlock();
    }
}
final E itemAt(int i) {
    
    return (E) items[i];
}

移除元素-boolean remove(Object o)

• 移除队列中与元素o相等【指的是equals方法判定相同】的元素，移除成功返回true，如果队列为空或没有匹配元素，则返回false。

    public boolean remove(Object o) {
    
        if (o == null) return false;
        //获取数组数据
        final Object[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();//加锁
        try {
    
            //如果此时队列不为null，这里是为了防止并发情况
            if (count > 0) {
    
                //获取下一个要添加元素时的索引
                final int putIndex = this.putIndex;
                //获取当前要被删除元素的索引
                int i = takeIndex;
                //执行循环查找要删除的元素
                do {
    
                     // 找到了对应的元素的位置，removeAt删除该位置的元素
                    if (o.equals(items[i])) {
    
                        removeAt(i);//执行删除
                        return true;//删除成功返回true
                    }
                    //当前删除索引执行加1后判断是否与数组长度相等
                    //若为true，说明索引已到数组尽头，将i设置为0
                    if (++i == items.length)
                        i = 0; 
                } while (i != putIndex);
                 //到达区间[takeIndex, putIndex)的边界，说明所有非null元素都找遍了
            }
            return false;//没有找到元素
        } finally {
    
            lock.unlock();//解锁
        }
    }

    //移除removeIndex位置的元素:根据索引删除元素，实际上是把删除索引之后的元素往前移动一个位置
    void removeAt(final int removeIndex) {
    
     final Object[] items = this.items;
     //如果刚好删除的是队首，那刚好是一个出队动作
      if (removeIndex == takeIndex) {
    
          //如果是直接删除
          items[takeIndex] = null;
          //当前队列头元素加1并判断是否与数组长度相等，若为true设置为0
          if (++takeIndex == items.length)
              takeIndex = 0;
          count--;//队列元素减1
          if (itrs != null)
              itrs.elementDequeued();//更新迭代器中的数据
      } 
      //其他情况
      else {
    
      //如果要删除的元素不在队列头部，
      //那么只需循环迭代把删除元素后面的所有元素往前移动一个位置
          //获取下一个要被添加的元素的索引，作为循环判断结束条件
          final int putIndex = this.putIndex;
          //执行循环
          for (int i = removeIndex;;) {
    
              //获取要删除节点索引的下一个索引
              int next = i + 1;
              //判断是否已为数组长度，如果是从数组头部（索引为0）开始找
              if (next == items.length)
                  next = 0;
               //如果查找的索引不等于要添加元素的索引，说明移除的不是队尾，后面的元素补充上来
              if (next != putIndex) {
    
                  items[i] = items[next];//把后一个元素前移覆盖要删除的元
                  i = next;
              } else {
    
                //在removeIndex索引之后的元素都往前移动完毕后清空最后一个元素
                  items[i] = null;
                  //最后putIndex当然也得左移，i此时肯定是putIndex - 1
                  this.putIndex = i;
                  break;//结束循环
              }
          }
          count--;//队列元素减1
          if (itrs != null)
              itrs.removedAt(removeIndex);//更新迭代器数据
      }
      notFull.signal();//唤醒添加线程
    }

九.总结

一波源码看下来，ArrayBlockingQueue 使用可重入锁 ReentrantLock保证线程安全，通过两个 Condition 实现生产者-消费者模型，看起来很简单的样子，这背后要感谢 ReentrantLock 和 Condition 的功劳！

1. ArrayBlockingQueue基于数组的有界阻塞队列，必须指定容量大小。阻塞式+超时机制的读写元素操作。
2. 通过ReentrantLock + Condition实现并发环境下的等待通知机制：读操作和写操作都需要获取到AQS独占锁才能进行操作如果队列为空，则读线程将会被包装为条件节点扔到读线程等待条件队列中阻塞，等待写线程写入新的元素，并唤醒等待中的读线程，反之亦然。
3. ArrayBlockingQueue利用了ReentrantLock来保证线程的安全性，针对队列的修改都需要加全局锁。在一般的应用场景下已经足够。对于超高并发的环境，由于生产者-消息者共用一把锁，可能出现性能瓶颈。

后续，我们会介绍另一种基于单链表实现的阻塞队列——LinkedBlockingQueue，该队列的最大特点是使用了“两把锁”，以提升吞吐量。