一、ArrayBlockingQueue

ArrayBlockingQueue 是一个用数组实现的有界阻塞队列，其内部按先进先出的原则对元素进行排序，其中put方法和take方法为添加和删除的阻塞方法，下面我们通过ArrayBlockingQueue队列实现一个生产者消费者的案例，通过该案例简单了解其使用方式

代码比较简单， Consumer 消费者和 Producer 生产者，通过ArrayBlockingQueue 队列获取和添加元素，其中消费者调用了take()方法获取元素当队列没有元素就阻塞，生产者调用put()方法添加元素，当队列满时就阻塞，通过这种方式便实现生产者消费者模式。比直接使用等待唤醒机制或者Condition条件队列来得更加简单。执行代码，打印部分Log如下

public class ArrayBlockingQueueDemo {private final static ArrayBlockingQueue<Apple> queue= new ArrayBlockingQueue<>(1);public static void main(String[] args){new Thread(new Producer(queue)).start();new Thread(new Producer(queue)).start();new Thread(new Consumer(queue)).start();new Thread(new Consumer(queue)).start();}
}class Apple {public Apple(){}}/*** 生产者线程*/
class Producer implements Runnable{private final ArrayBlockingQueue<Apple> mAbq;Producer(ArrayBlockingQueue<Apple> arrayBlockingQueue){this.mAbq = arrayBlockingQueue;}@Overridepublic void run() {while (true) {Produce();}}private void Produce(){try {Apple apple = new Apple();mAbq.put(apple);System.out.println("生产:"+apple);} catch (InterruptedException e) {e.printStackTrace();}}
}/*** 消费者线程*/
class Consumer implements Runnable{private ArrayBlockingQueue<Apple> mAbq;Consumer(ArrayBlockingQueue<Apple> arrayBlockingQueue){this.mAbq = arrayBlockingQueue;}@Overridepublic void run() {while (true){try {TimeUnit.MILLISECONDS.sleep(1000);comsume();} catch (InterruptedException e) {e.printStackTrace();}}}private void comsume() throws InterruptedException {Apple apple = mAbq.take();System.out.println("消费Apple="+apple);}
}

生产:com.ee.Apple@421674f0
消费Apple=com.ee.Apple@421674f0
消费Apple=com.ee.Apple@47c48106
消费Apple=com.ee.Apple@4cd182c2
生产:com.ee.Apple@4cd182c2
生产:com.ee.Apple@47c48106
生产:com.ee.Apple@48ee1f80
消费Apple=com.ee.Apple@48ee1f80 

有点需要注意的是ArrayBlockingQueue内部的阻塞队列是通过重入锁ReenterLock和Condition条件队列实现的，所以ArrayBlockingQueue中的元素存在公平访问与非公平访问的区别，对于公平访问队列，被阻塞的线程可以按照阻塞的先后顺序访问队列，即先阻塞的线程先访问队列。而非公平队列，当队列可用时，阻塞的线程将进入争夺访问资源的竞争中，也就是说谁先抢到谁就执行，没有固定的先后顺序。创建公平与非公平阻塞队列代码如下： 

//默认非公平阻塞队列
ArrayBlockingQueue queue = new ArrayBlockingQueue(2);
//公平阻塞队列
ArrayBlockingQueue queue1 = new ArrayBlockingQueue(2,true);//构造方法源码
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();}

其他方法如下：

//自动移除此队列中的所有元素。
void clear() //如果此队列包含指定的元素，则返回 true。          
boolean contains(Object o) //移除此队列中所有可用的元素，并将它们添加到给定collection中。           
int drainTo(Collection<? super E> c) //最多从此队列中移除给定数量的可用元素，并将这些元素添加到给定collection 中。       
int drainTo(Collection<? super E> c, int maxElements) //返回在此队列中的元素上按适当顺序进行迭代的迭代器。         
Iterator<E> iterator() //返回队列还能添加元素的数量
int remainingCapacity() //返回此队列中元素的数量。      
int size() //返回一个按适当顺序包含此队列中所有元素的数组。
Object[] toArray() //返回一个按适当顺序包含此队列中所有元素的数组；返回数组的运行时类型是指定数组的运行时类型。      
<T> T[] toArray(T[] a)

ArrayBlockingQueue原理概要

ArrayBlockingQueue的内部是通过一个可重入锁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, or add 方法*/int putIndex;/** 队列元素的个数 */int count;/** 控制并非访问的锁 */final ReentrantLock lock;/**notEmpty条件对象，用于通知take方法队列已有元素，可执行获取操作 */private final Condition notEmpty;/**notFull条件对象，用于通知put方法队列未满，可执行添加操作 */private final Condition notFull;/**迭代器*/transient Itrs itrs = null;}

从成员变量可看出，ArrayBlockingQueue内部确实是通过数组对象items来存储所有的数据，值得注意的是ArrayBlockingQueue通过一个ReentrantLock来同时控制添加线程与移除线程的并发访问，这点与LinkedBlockingQueue区别很大(稍后会分析)。而对于notEmpty条件对象则是用于存放等待或唤醒调用take方法的线程，告诉他们队列已有元素，可以执行获取操作。同理notFull条件对象是用于等待或唤醒调用put方法的线程，告诉它们，队列未满，可以执行添加元素的操作。takeIndex代表的是下一个方法(take，poll，peek，remove)被调用时获取数组元素的索引，putIndex则代表下一个方法（put, offer, or add）被调用时元素添加到数组中的索引。图示如下

 ArrayBlockingQueue的(阻塞)添加的实现原理 

//add方法实现，间接调用了offer(e)
public boolean add(E e) {if (offer(e))return true;elsethrow new IllegalStateException("Queue full");}//offer方法
public boolean offer(E e) {checkNotNull(e);//检查元素是否为nullfinal ReentrantLock lock = this.lock;lock.lock();//加锁try {if (count == items.length)//判断队列是否满return false;else {enqueue(e);//添加元素到队列return true;}} finally {lock.unlock();}}//入队操作
private void enqueue(E x) {//获取当前数组final Object[] items = this.items;//通过putIndex索引对数组进行赋值items[putIndex] = x;//索引自增，如果已是最后一个位置，重新设置 putIndex = 0;if (++putIndex == items.length)putIndex = 0;count++;//队列中元素数量加1//唤醒调用take()方法的线程，执行元素获取操作。notEmpty.signal();
}

这里的add方法和offer方法实现比较简单，其中需要注意的是enqueue(E x)方法，其方法内部通过putIndex索引直接将元素添加到数组items中，这里可能会疑惑的是当putIndex索引大小等于数组长度时，需要将putIndex重新设置为0，这是因为队列执行元素获取时总是从队列头部获取，而添加元素从队列尾部获取所以当队列索引（从0开始）与数组长度相等时，下次我们就需要从数组头部开始添加了，如下图演示

 接着看put方法，它是一个阻塞添加的方法

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

put方法是一个阻塞的方法，如果队列元素已满，那么当前线程将会被notFull条件对象挂起加到等待队列中，直到队列有空档才会唤醒执行添加操作。但如果队列没有满，那么就直接调用enqueue(e)方法将元素加入到数组队列中。到此我们对三个添加方法即put，offer，add都分析完毕，其中offer，add在正常情况下都是无阻塞的添加，而put方法是阻塞添加。这就是阻塞队列的添加过程。说白了就是当队列满时通过条件对象Condtion来阻塞当前调用put方法的线程，直到线程又再次被唤醒执行。总得来说添加线程的执行存在以下两种情况，一是，队列已满，那么新到来的put线程将添加到notFull的条件队列中等待，二是，有移除线程执行移除操作，移除成功同时唤醒put线程，如下图所示

ArrayBlockingQueue的(阻塞)移除实现原理 

关于删除先看poll方法，该方法获取并移除此队列的头元素，若队列为空，则返回 null

public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {//判断队列是否为null，不为null执行dequeue()方法，否则返回nullreturn (count == 0) ? null : dequeue();} finally {lock.unlock();}}//删除队列头元素并返回private E dequeue() {//拿到当前数组的数据final Object[] items = this.items;@SuppressWarnings("unchecked")//获取要删除的对象E x = (E) items[takeIndex];将数组中takeIndex索引位置设置为nullitems[takeIndex] = null;//takeIndex索引加1并判断是否与数组长度相等，//如果相等说明已到尽头，恢复为0if (++takeIndex == items.length)takeIndex = 0;count--;//队列个数减1if (itrs != null)itrs.elementDequeued();//同时更新迭代器中的元素数据//删除了元素说明队列有空位，唤醒notFull条件对象添加线程，执行添加操作notFull.signal();return x;}

poll()，获取并删除队列头元素，队列没有数据就返回null，内部通过dequeue()方法删除头元素，注释很清晰，这里不重复了。接着看remove(Object o)方法

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 {//找到要删除的元素if (o.equals(items[i])) {removeAt(i);//执行删除return true;//删除成功返回true}//当前删除索引执行加1后判断是否与数组长度相等//若为true，说明索引已到数组尽头，将i设置为0if (++i == items.length)i = 0; } while (i != putIndex);//继承查找}return false;} finally {lock.unlock();}
}//根据索引删除元素，实际上是把删除索引之后的元素往前移动一个位置
void removeAt(final int removeIndex) {final Object[] items = this.items;//先判断要删除的元素是否为当前队列头元素if (removeIndex == takeIndex) {//如果是直接删除items[takeIndex] = null;//当前队列头元素加1并判断是否与数组长度相等，若为true设置为0if (++takeIndex == items.length)takeIndex = 0;count--;//队列元素减1if (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;this.putIndex = i;break;//结束循环}}count--;//队列元素减1if (itrs != null)itrs.removedAt(removeIndex);//更新迭代器数据}notFull.signal();//唤醒添加线程}

remove(Object o)方法的删除过程相对复杂些，因为该方法并不是直接从队列头部删除元素。首先线程先获取锁，再一步判断队列count>0,这点是保证并发情况下删除操作安全执行。接着获取下一个要添加源的索引putIndex以及takeIndex索引 ，作为后续循环的结束判断，因为只要putIndex与takeIndex不相等就说明队列没有结束。然后通过while循环找到要删除的元素索引，执行removeAt(i)方法删除，在removeAt(i)方法中实际上做了两件事，一是首先判断队列头部元素是否为删除元素，如果是直接删除，并唤醒添加线程，二是如果要删除的元素并不是队列头元素，那么执行循环操作，从要删除元素的索引removeIndex之后的元素都往前移动一个位置，那么要删除的元素就被removeIndex之后的元素替换，从而也就完成了删除操作。

接着看take()方法，是一个阻塞方法，直接获取队列头元素并删除。

//从队列头部删除，队列没有元素就阻塞，可中断public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();//中断try {//如果队列没有元素while (count == 0)//执行阻塞操作notEmpty.await();return dequeue();//如果队列有元素执行删除操作} finally {lock.unlock();}}

take方法其实很简单，有就删除没有就阻塞，注意这个阻塞是可以中断的，如果队列没有数据那么就加入notEmpty条件队列等待(有数据就直接取走，方法结束)，如果有新的put线程添加了数据，那么put操作将会唤醒take线程，执行take操作。图示如下

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

peek方法非常简单，直接返回当前队列的头元素但不删除任何元素。ok~，到此对于ArrayBlockingQueue的主要方法就分析完了。