网站建设好后怎么制作网页/外呼系统电销

1、简介

ArrayBlockingQueue，顾名思义：基于数组的阻塞队列。数组是要指定长度的，所以使用ArrayBlockingQueue时必须指定长度，也就是它是一个有界队列。

它实现了BlockingQueue接口，有着队列、集合以及阻塞队列的所有方法，队列类图如下图所示：

既然它在JUC包内，说明使用它是线程安全的，它内部使用ReentrantLock来保证线程安全。ArrayBlockingQueue支持对生产者线程和消费者线程进行公平的调度，默认情况下是不保证公平性的。公平性通常会降低吞吐量，但是减少了可变性和避免了线程饥饿问题。

JUC包指java.util.concurrent目录下的类

2、数据结构

通常，队列的实现方式有数组和链表两种方式。对于数组这种实现方式来说，我们可以通过维护一个队尾指针，使得在入队的时候可以在O(1)的时间内完成；但是对于出队操作，在删除队头元素之后，必须将数组中的所有元素都往前移动一个位置，这个操作的复杂度达到了O(n)，效果并不是很好。如下图所示：

为了解决这个问题，我们可以使用另外一种逻辑结构来处理数组中各个位置之间的关系。假设现在我们有一个数组A[1…n]，我们可以把它想象成一个环型结构，即A[n]之后是A[1]，相信了解过一致性Hash算法的童鞋应该很容易能够理解。如下图所示：我们可以使用两个指针，分别维护队头和队尾两个位置，使入队和出队操作都可以在O(1)的时间内完成。当然，这个环形结构只是逻辑上的结构，实际的物理结构还是一个普通的数据。

因此ArrayBlockingQueue的实现是一个循环数组，使用takeIndex和putIndex来控制元素的出入队列，效率高。

讲完ArrayBlockingQueue的数据结构，接下来我们从源码层面看看它是如何实现阻塞的。

3、源码分析

3.1、属性

JDK1.8

public class ArrayBlockingQueue {final Object[] items;   //队列的底层结构int takeIndex;    //队头指针int putIndex;     //队尾指针int count;       //队列中的元素个数final ReentrantLock lock;//并发时的两种状态private final Condition notEmpty;private final Condition notFull;

items是一个数组，用来存放入队的数据，count表示队列中元素的个数。takeIndex和putIndex分别代表队头和队尾指针。

说明：Lock的作用是提供独占锁机制，来保护竞争的资源；而Condition是为了更精细的对锁进行控制，但是依赖于lock，通过某个条件对多线程进行控制。

• notEmpty表示"锁的非空条件"。当某线程想从队列中获取数据的时候，而此时队列中的数据为空，则该线程通过notEmpty.await()方法进行等待；当其他线程向队列中插入元素之后，就调用notEmpty.signal()方法进行唤醒之前等待的线程。

• 同理，notFull表示“锁满的条件“。当某个线程向队列中插入元素，而此时队列已满时，该线程等待，即阻塞通过notFull.wait()方法；其他线程从队列中取出元素之后，就唤醒该等待的线程，这个线程调用notFull.signal()方法。

3.2、构造函数

public class ArrayBlockingQueue {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();}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 exclusiontry {int i = 0;try {for (E e : c) {checkNotNull(e);items[i++] = e;}} catch (ArrayIndexOutOfBoundsException ex) {throw new IllegalArgumentException();}count = i;putIndex = (i == capacity) ? 0 : i;} finally {lock.unlock();}}

• 第一个构造函数只需要制定队列大小，默认为非公平锁。

• 第二个构造函数可以手动制定公平性和队列大小。

• 第三个构造函数里面使用了ReentrantLock来加锁，然后把传入的集合元素按顺序一个个放入items中。这里加锁目的不是使用它的互斥性，而是让items中的元素对其他线程可见（用的是AQS里的state的volatile可见性）。

3.3、方法

3.3.1、入队方法

ArrayBlockingQueue 提供了多种入队操作的实现来满足不同情况下的需求，入队操作有如下几种：

• boolean add(E e);
• void put(E e)； //阻塞，其余非阻塞
• boolean offer(E e)；
• boolean offer(E e, long timeout, TimeUnit unit)。

add(E e)

public boolean add(E e) {   // ArrayBlockingQueue.javareturn super.add(e);
}
}//super.add(e)
public boolean add(E e) {    // AbstractQueue.javaif (offer(e))            //复用offer方法return true;elsethrow new IllegalStateException("Queue full");   //抛出异常
}

可以看到add方法调用的是父类，也就是AbstractQueue的add方法，它实际上调用的就是offer方法，并进行封装，针对返回值false情况抛出异常。

offer(E e)

我们接着上面的add方法来看offer方法：

public boolean offer(E e) {    //ArrayBlockingQueue.javacheckNotNull(e);final ReentrantLock lock = this.lock;lock.lock();try {if (count == items.length)    //如果相等，则说明队列满return false;else {enqueue(e);return true;}} finally {lock.unlock();}
}

offer方法在队列满了的时候返回false，否则调用enqueue方法插入元素，并返回true。

private void enqueue(E x) {final Object[] items = this.items;items[putIndex] = x;   //存放当前元素// 圆环的index操作if (++putIndex == items.length)putIndex = 0;    //putIndex 标记队尾，下一个元素可以存放的位置count++;             //数组内实际元素个数+1，count用来判断满队列或空队列notEmpty.signal();   //唤醒等待获取元素的线程
}

enqueue方法首先把元素放在items的putIndex位置，接着判断在putIndex+1等于队列的长度时把putIndex设置为0，也就是上面提到的圆环的index操作。最后唤醒等待获取元素的线程。

从enqueue方法，可以得知采用了圆环操作，可以与后面的dequeue方法对比来看。圆环操作的精髓就是当添加元素到队列最后一个位置时，重新从队列头开始循环，通过内部的putIndex 指针实现。

offer(E e, long timeout, TimeUnit unit)

offer(E e, long timeout, TimeUnit unit)方法只是在offer(E e)的基础上增加了超时时间的概念。

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循环的目的是防止在中断后没有到达传入的timeout时间，继续重试while (count == items.length) {if (nanos <= 0)return false;    //超时// 等待nanos纳秒，返回剩余的等待时间(可被中断)nanos = notFull.awaitNanos(nanos);}enqueue(e);   //没有超时，继续插入数据return true;} finally {lock.unlock();}
}

该方法利用了Condition的awaitNanos方法，等待指定时间，因为该方法可中断，所以这里利用while循环来处理中断后还有剩余时间的问题，等待时间到了以后调用enqueue方法放入队列。

Condition的awaitNanos方法返回值是被唤醒后剩余的时间，比如我预期等待1000ms，然后等待了200ms，那么返回值是800，说明没有超时；如果返回值<=0，说明超时了。

put(E e)

阻塞，直到队列有空位产生

public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == items.length)notFull.await();   //队列满了，持续等待enqueue(e);} finally {lock.unlock();}
}

put方法在count等于items长度时，一直等待，直到被其他线程唤醒。唤醒后调用enqueue方法放入队列。

3.3.2、出队方法

入队列的方法说完后，我们来说说出队列的方法。ArrayBlockingQueue提供了多种出队操作的实现来满足不同情况下的需求，如下：

• E poll();
• E poll(long timeout, TimeUnit unit);
• E take()。

poll()

public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {return (count == 0) ? null : dequeue();  //队列为空，返回null} finally {lock.unlock();}
}

poll方法是非阻塞方法，如果队列没有元素返回null，否则调用dequeue把队首的元素出队列。

private E dequeue() {final Object[] items = this.items;@SuppressWarnings("unchecked")E x = (E) items[takeIndex];items[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;     //圆环原理，循环count--;         //队列中的实际元素个数-1if (itrs != null)itrs.elementDequeued();notFull.signal();return x;
}

dequeue会根据takeIndex获取到该位置的元素，并把该位置置为null，接着利用圆环原理，在takeIndex到达列表长度时设置为0，最后唤醒等待元素放入队列的线程。

poll(long timeout, TimeUnit unit)

该方法是poll()的可配置超时等待方法，和上面的offer一样，使用while循环和Condition的awaitNanos来进行等待，等待时间到后执行dequeue获取元素。

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

take()

队列为空就阻塞

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)notEmpty.await();  //阻塞return dequeue();} finally {lock.unlock();}
}

3.3.3、获取元素方法

peek()

查询元素，不会从队列中删除。

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];
}

这里获取元素时上锁是为了避免脏数据的产生。

3.3.4、删除元素方法

remove(Object o)

删除指定对象，注意与 remove()无参的区别，后者是删除队首的元素。

我们可以想象一下，队列中删除某一个元素时，是不是要遍历整个数据找到该元素，并把该元素后的所有元素往前移一位？队列也一样， 只不过遍历的时候，起始点和结束点有区别。

该方法的时间复杂度为O(n)。

public boolean remove(Object o) {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;      //下标// 从takeIndex一直遍历到putIndex，直到找到和元素o相同的元素，调用removeAt进行删除do {if (o.equals(items[i])) {removeAt(i);return true;}if (++i == items.length)i = 0;    // 环形结构，队尾需要重置到队头} while (i != putIndex);}return false;} finally {lock.unlock();}
}

remove方法比较简单，它从takeIndex一直遍历到putIndex，直到找到和元素o相同的元素，调用removeAt进行删除。我们重点来看一下removeAt方法。

void removeAt(final int removeIndex) {final Object[] items = this.items;if (removeIndex == takeIndex) {        //刚好是队尾元素，直接删，不需要移位// removing front item; just advanceitems[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();} else {// an "interior" remove// slide over all others up through putIndex.final int putIndex = this.putIndex;   //队尾，需要把待删除元素后至队尾的所有元素前移for (int i = removeIndex;;) {int next = i + 1;if (next == items.length)next = 0;     //环形原理，需要找到队头if (next != putIndex) {items[i] = items[next];  //非队尾元素，前移一位i = next;} else {items[i] = null;   //原来队尾元素置空，并标记为下次插入元素的下标this.putIndex = i;break;}}count--;if (itrs != null)itrs.removedAt(removeIndex);}notFull.signal();
}

removeAt的处理方式和我想的稍微有一点出入，它内部分为两种情况来考虑

• removeIndex == takeIndex
• removeIndex != takeIndex

也就是我考虑的时候没有考虑边界问题。当removeIndex == takeIndex时就不需要后面的元素整体往前移了，而只需要把takeIndex的指向下一个元素即可（还记得前面说的ArrayBlockingQueue可以类比为圆环吗）。

当removeIndex != takeIndex时，通过putIndex将removeIndex后的元素往前移一位。

4、总结

ArrayBlockingQueue是一个阻塞队列，内部由ReentrantLock来实现线程安全，由Condition的await和signal来实现等待唤醒的功能。

它的数据结构是数组，准确的说是一个循环数组（可以类比一个圆环），所有的下标在到达最大长度时自动从0继续开始。

参考：
ArrayBlockingQueue
《JUC回顾之-ArrayBlockingQueue底层实现和原理》