一.概述
ArrayBlockingQueue是一个由数组支持的有界阻塞队列
1. 队列元素是先进先出（FIFO）
2. 队列的头元素是在队列中时间最长的元素，因为最先入队列的，获取操作(poll、peek、take)返回头元素
3. 队列的尾元素是在队列中时间最短的元素,因为最后入队列的，新元素插入(offer、put)队列的尾部
4. 数组大小一旦固定，就不允许再增加变量， 试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞
5. 插入的元素不能为null，否则会抛出空指针
6. ArrayBlockingQueue只有一个锁对象lock，控制着入队列和出队列，一次只能执行其一，但LinkedBlockingQueue有两个锁对象，分别控制着入队列和出队列，二者可以同时进行，通过原子性count解决对同一个变量进行并发修改的线程安全
7. ArrayBlockingQueue因为只有一个锁对象，所以用int型count记录元素数量，不用使用原子性int
二.ArrayblockingQueue对象结构
public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable
三.对象成员变量
// 队列数组
final Object[] items;
// 出队列索引
int takeIndex;
// 入队列索引
int putIndex;
// 队列中元素总数
int count;
// 可重入锁
final ReentrantLock lock;
/** * 队列非空监听器 * 1. 队列为空，插入一个元素出队列一个元素，notEmpty.await()，消费线程等待 * 2. 队列插入一个元素，notEmpty.signal()，唤醒等待的消费线程 * */
private final Condition notEmpty;
/** * 队列未满监听器 * 1. 队列已满，插入一个元素，notFull.await()，生产线程等待 * 2. 出队列一个元素，notFull.signal()，唤醒等待的生产线程 * */
private final Condition notFull;
transient Itrs itrs = null;四.创建对象实例
/** * 根据指定容量大小和默认非公平锁创建ArrayBlockingQueue实例 * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity < 1} */
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
/** * 根据指定的容量大小和指定的锁访问策略创建一个ArrayBlockingQueue实例 * * @param capacity the capacity of this queue * @param fair true表示公平锁，false表示非公平锁 * @throws IllegalArgumentException if {@code capacity < 1} */
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();
}
/** * @param capacity the capacity of this queue * @param fair * @param c the collection of elements to initially contain * @throws IllegalArgumentException if {@code capacity} is less than * {@code c.size()}, or less than 1. * @throws NullPointerException if the specified collection or any * of its elements are null */
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);
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock();
}
}五.入队列
5.1 入队列方法
/** * 在putIndex插入元素x，只有在获取锁时才会被调用 */
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
// 插入完成，putIndex+1,表示下一个插入的索引
// 如果当前插入数组索引最大处，下一个从索引0处插入
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}5.2 入队列步骤
5.2.1 空队列
一个大小为5的数组，入队列索引putIndex和出队列takeIndex都为0
5.2.2 入队列1
插入一个元素，putIndex加1
5.2.3 满队列
- 队列满了，入队列操作会阻塞
- 如果当前putIndex为数组尾，加1后为0，即下一个从数组头看似是插入
5.2.4 出队列1
5.2.5 出队列2
5.2.6 入队列1
5.3 入队列方法
5.3.1 add(E e)
- 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true
- 如果队列已满，抛出异常IllegalStateException
/** * @throws IllegalStateException 如果队列已满 * @throws NullPointerException 待插入元素为null * */
public boolean add(E e) {
return super.add(e);
}
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}5.3.2 offer(E e)
- 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true
- 如果队列已满，返回false
- offer(e)通常要优于add(e),因为add(e)在队列满了时会抛出IllegalStateExcepti
// @throws NullPointerException 待插入元素为null
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}5.3.3 offer(E e, long timeout, TimeUnit unit)
- 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true
- 如果队列已满，等待指定的时间timeout(单位为unit)以使空间变为可用
// @throws NullPointerException 待插入元素为null
// @throws InterruptedException 在等待时被中断
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) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}5.3.4 put(E e)
- 如果队列未满，立即执行在队列的尾部插入指定的元素e
- 如果队列已满，等待队列空间可用后唤醒再插入
// @throws NullPointerException 待插入元素为null
// @throws InterruptedException 如果在等待时被中断
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();
}
}六.出队列
6.1 出队列方法
从索引takeIndex处获取并移除元素，只有在获取锁时才会被调用
/** * 从索引takeIndex处获取并移除元素，只有在获取锁时才会被调用 */
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
// 获取并移除索引takeIndex处的索引后，加1
// 如果当前获取元素位于数组索引最大处，下一次获取索引从0开始
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}6.2 出队列步骤
参考标题5中出入队列步骤
6.3 出队列方法
6.3.1 peek()
获取但不移除此队列的头；如果此队列为空，则返回 null
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
@SuppressWarnings("unchecked")
final E itemAt(int i) {
return (E) items[i];
}6.3.2 poll()
获取并移除此队列的头，如果此队列为空，则返回 null
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}6.3.3 poll(long timeout, TimeUnit unit)
获取并移除此队列的头部，在指定的等待时间前等待可用的元素（如果有必要）
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();
}
}6.3.4 take()
获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}七.移除指定元素
- 获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）
- 如果待移除元素索引等于putIndex，等同于poll()，直接移除
- 如果待移除元素索引removeIndex不等于putIndex,从removeIndex处开始向后遍历，将待移除元素之后的元素向前移动一个
void removeAt(final int removeIndex) {
final Object[] items = this.items;
if (removeIndex == takeIndex) {
// removing front item; just advance
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
} else {
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();
}八.总结
- ArrayBlockingQueue由于其底层基于数组，并且在创建时指定存储的大小，在完成后就会立即在内存分配固定大小容量的数组元素，因此其存储通常有限，故其是一个“有界“的阻塞队列
- 由于ArrayBlockingQueue的有界性，因此其能够更好的对于性能进行预测，LinkedBlockingQueue由于没有限制大小，当任务非常多的时候，不停地向队列中存储，就有可能导致内存溢出的情况发生
- ArrayBlockingQueue中在入队列和出队列操作过程中，使用的是同一个lock，所以即使在多核CPU的情况下，其读取和操作的都无法做到并行
- LinkedBlockingQueue的读取和插入操作所使用的锁是两个不同的lock，它们之间的操作互相不受干扰，因此两种操作可以并行完成，故LinkedBlockingQueue的吞吐量要高于ArrayBlockingQueue
九.参考
https://www.cnblogs.com/skywang12345/p/3498652.html
十.源码
package java.util.concurrent;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.lang.ref.WeakReference;
import java.util.Spliterators;
import java.util.Spliterator;
/** * 一个由数组支持的有界阻塞队列。 * 1. 此队列按 FIFO（先进先出）原则对元素进行排序。 * 2. 队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。 * 3. 新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。 * 4. 这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。 * 试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。 * 5. 此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。默认情况下，不保证是这种排序。然而，通过将公平性 (fairness) 设置为 true 而构造的队列允许按照 FIFO 顺序访问线程。公平性通常会降低吞吐量，但也减少了可变性和避免了“不平衡性”。 * * @since 1.5 * @author Doug Lea * @param <E> the type of elements held in this collection */
public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
private static final long serialVersionUID = -817911632652898426L;
// 队列数组
final Object[] items;
// 出队列索引
int takeIndex;
// 入队列索引
int putIndex;
// 队列中元素总数
int count;
// 可重入锁
final ReentrantLock lock;
/** * 队列非空监听器 * 1. 队列为空，插入一个元素出队列一个元素，notEmpty.await()，消费线程等待 * 2. 队列插入一个元素，notEmpty.signal()，唤醒等待的消费线程 * */
private final Condition notEmpty;
/** * 队列未满监听器 * 1. 队列已满，插入一个元素，notFull.await()，生产线程等待 * 2. 出队列一个元素，notFull.signal()，唤醒等待的生产线程 * */
private final Condition notFull;
transient Itrs itrs = null;
/** * 根据指定容量大小和默认非公平锁创建ArrayBlockingQueue实例 * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity < 1} */
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
/** * 根据指定的容量大小和指定的锁访问策略创建一个ArrayBlockingQueue实例 * * @param capacity the capacity of this queue * @param fair true表示公平锁，false表示非公平锁 * @throws IllegalArgumentException if {@code capacity < 1} */
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();
}
/** * @param capacity the capacity of this queue * @param fair * @param c the collection of elements to initially contain * @throws IllegalArgumentException if {@code capacity} is less than * {@code c.size()}, or less than 1. * @throws NullPointerException if the specified collection or any * of its elements are null */
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);
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock();
}
}
/** * Circularly decrement i. */
final int dec(int i) {
return ((i == 0) ? items.length : i) - 1;
}
/** * 检查插入的对象是否为空，为空抛出空指针 * * @param v the element */
private static void checkNotNull(Object v) {
if (v == null)
throw new NullPointerException();
}
/** * 在putIndex插入元素x，只有在获取锁时才会被调用 */
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
// 插入完成，putIndex+1,表示下一个插入的索引
// 如果当前插入数组索引最大处，下一个从索引0处插入
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
/** * 从索引takeIndex处获取并移除元素，只有在获取锁时才会被调用 */
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
// 获取并移除索引takeIndex处的索引后，加1
// 如果当前获取元素位于数组索引最大处，下一次获取索引从0开始
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}
/** * add(e)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true * 2. 如果队列已满，抛出异常IllegalStateException * * @throws NullPointerException 待插入元素为null * @throws IllegalStateException 如果队列已满时执行插入 * */
public boolean add(E e) {
return super.add(e);
}
/** * offer(e)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true * 2. 如果队列已满，返回false * * offer(e)通常要优于add(e),因为add(e)在队列满了时会抛出IllegalStateException异常 * * @throws NullPointerException 待插入元素为null * */
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}
/** * offer(E e, long timeout, TimeUnit unit)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e，成功返回true * 2. 如果队列已满，等待指定的时间timeout(单位为unit)以使空间变为可用。 * * * @throws NullPointerException 待插入元素为null * @throws InterruptedException 在等待时被中断 * */
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) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
/** * put(e)测试： * 1. 如果队列未满，立即执行在队列的尾部插入指定的元素e * 2. 如果队列已满，等待队列空间可用后唤醒再插入 * * @throws NullPointerException 待插入元素为null * @throws InterruptedException 如果在等待时被中断 * */
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();
}
}
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
@SuppressWarnings("unchecked")
final E itemAt(int i) {
return (E) items[i];
}
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
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();
}
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
/** * Deletes item at array index removeIndex. * Utility for remove(Object) and iterator.remove. * Call only when holding lock. */
void removeAt(final int removeIndex) {
final Object[] items = this.items;
if (removeIndex == takeIndex) {
// removing front item; just advance
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
} else {
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();
}
// this doc comment is overridden to remove the reference to collections
// greater in size than Integer.MAX_VALUE
/** * Returns the number of elements in this queue. * * @return the number of elements in this queue */
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
// this doc comment is a modified copy of the inherited doc comment,
// without the reference to unlimited queues.
/** * Returns the number of additional elements that this queue can ideally * (in the absence of memory or resource constraints) accept without * blocking. This is always equal to the initial capacity of this queue * less the current {@code size} of this queue. * * <p>Note that you <em>cannot</em> always tell if an attempt to insert * an element will succeed by inspecting {@code remainingCapacity} * because it may be the case that another thread is about to * insert or remove an element. */
public int remainingCapacity() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return items.length - count;
} finally {
lock.unlock();
}
}
/** * Removes a single instance of the specified element from this queue, * if it is present. More formally, removes an element {@code e} such * that {@code o.equals(e)}, if this queue contains one or more such * elements. * Returns {@code true} if this queue contained the specified element * (or equivalently, if this queue changed as a result of the call). * * <p>Removal of interior elements in circular array based queues * is an intrinsically slow and disruptive operation, so should * be undertaken only in exceptional circumstances, ideally * only when the queue is known not to be accessible by other * threads. * * @param o element to be removed from this queue, if present * @return {@code true} if this queue changed as a result of the call */
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;
do {
if (o.equals(items[i])) {
removeAt(i);
return true;
}
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
return false;
} finally {
lock.unlock();
}
}
/** * Returns {@code true} if this queue contains the specified element. * More formally, returns {@code true} if and only if this queue contains * at least one element {@code e} such that {@code o.equals(e)}. * * @param o object to be checked for containment in this queue * @return {@code true} if this queue contains the specified element */
public boolean contains(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;
do {
if (o.equals(items[i]))
return true;
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
return false;
} finally {
lock.unlock();
}
}
/** * Returns an array containing all of the elements in this queue, in * proper sequence. * * <p>The returned array will be "safe" in that no references to it are * maintained by this queue. (In other words, this method must allocate * a new array). The caller is thus free to modify the returned array. * * <p>This method acts as bridge between array-based and collection-based * APIs. * * @return an array containing all of the elements in this queue */
public Object[] toArray() {
Object[] a;
final ReentrantLock lock = this.lock;
lock.lock();
try {
final int count = this.count;
a = new Object[count];
int n = items.length - takeIndex;
if (count <= n)
System.arraycopy(items, takeIndex, a, 0, count);
else {
System.arraycopy(items, takeIndex, a, 0, n);
System.arraycopy(items, 0, a, n, count - n);
}
} finally {
lock.unlock();
}
return a;
}
/** * Returns an array containing all of the elements in this queue, in * proper sequence; the runtime type of the returned array is that of * the specified array. If the queue fits in the specified array, it * is returned therein. Otherwise, a new array is allocated with the * runtime type of the specified array and the size of this queue. * * <p>If this queue fits in the specified array with room to spare * (i.e., the array has more elements than this queue), the element in * the array immediately following the end of the queue is set to * {@code null}. * * <p>Like the {@link #toArray()} method, this method acts as bridge between * array-based and collection-based APIs. Further, this method allows * precise control over the runtime type of the output array, and may, * under certain circumstances, be used to save allocation costs. * * <p>Suppose {@code x} is a queue known to contain only strings. * The following code can be used to dump the queue into a newly * allocated array of {@code String}: * * <pre> {@code String[] y = x.toArray(new String[0]);}</pre> * * Note that {@code toArray(new Object[0])} is identical in function to * {@code toArray()}. * * @param a the array into which the elements of the queue are to * be stored, if it is big enough; otherwise, a new array of the * same runtime type is allocated for this purpose * @return an array containing all of the elements in this queue * @throws ArrayStoreException if the runtime type of the specified array * is not a supertype of the runtime type of every element in * this queue * @throws NullPointerException if the specified array is null */
@SuppressWarnings("unchecked")
public <T> T[] toArray(T[] a) {
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
final int count = this.count;
final int len = a.length;
if (len < count)
a = (T[])java.lang.reflect.Array.newInstance(
a.getClass().getComponentType(), count);
int n = items.length - takeIndex;
if (count <= n)
System.arraycopy(items, takeIndex, a, 0, count);
else {
System.arraycopy(items, takeIndex, a, 0, n);
System.arraycopy(items, 0, a, n, count - n);
}
if (len > count)
a[count] = null;
} finally {
lock.unlock();
}
return a;
}
public String toString() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
int k = count;
if (k == 0)
return "[]";
final Object[] items = this.items;
StringBuilder sb = new StringBuilder();
sb.append('[');
for (int i = takeIndex; ; ) {
Object e = items[i];
sb.append(e == this ? "(this Collection)" : e);
if (--k == 0)
return sb.append(']').toString();
sb.append(',').append(' ');
if (++i == items.length)
i = 0;
}
} finally {
lock.unlock();
}
}
/** * Atomically removes all of the elements from this queue. * The queue will be empty after this call returns. */
public void clear() {
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int k = count;
if (k > 0) {
final int putIndex = this.putIndex;
int i = takeIndex;
do {
items[i] = null;
if (++i == items.length)
i = 0;
} while (i != putIndex);
takeIndex = putIndex;
count = 0;
if (itrs != null)
itrs.queueIsEmpty();
for (; k > 0 && lock.hasWaiters(notFull); k--)
notFull.signal();
}
} finally {
lock.unlock();
}
}
/** * @throws UnsupportedOperationException {@inheritDoc} * @throws ClassCastException {@inheritDoc} * @throws NullPointerException {@inheritDoc} * @throws IllegalArgumentException {@inheritDoc} */
public int drainTo(Collection<? super E> c) {
return drainTo(c, Integer.MAX_VALUE);
}
/** * @throws UnsupportedOperationException {@inheritDoc} * @throws ClassCastException {@inheritDoc} * @throws NullPointerException {@inheritDoc} * @throws IllegalArgumentException {@inheritDoc} */
public int drainTo(Collection<? super E> c, int maxElements) {
checkNotNull(c);
if (c == this)
throw new IllegalArgumentException();
if (maxElements <= 0)
return 0;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int n = Math.min(maxElements, count);
int take = takeIndex;
int i = 0;
try {
while (i < n) {
@SuppressWarnings("unchecked")
E x = (E) items[take];
c.add(x);
items[take] = null;
if (++take == items.length)
take = 0;
i++;
}
return n;
} finally {
// Restore invariants even if c.add() threw
if (i > 0) {
count -= i;
takeIndex = take;
if (itrs != null) {
if (count == 0)
itrs.queueIsEmpty();
else if (i > take)
itrs.takeIndexWrapped();
}
for (; i > 0 && lock.hasWaiters(notFull); i--)
notFull.signal();
}
}
} finally {
lock.unlock();
}
}
/** * Returns an iterator over the elements in this queue in proper sequence. * The elements will be returned in order from first (head) to last (tail). * * <p>The returned iterator is * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>. * * @return an iterator over the elements in this queue in proper sequence */
public Iterator<E> iterator() {
return new Itr();
}
/** * Shared data between iterators and their queue, allowing queue * modifications to update iterators when elements are removed. * * This adds a lot of complexity for the sake of correctly * handling some uncommon operations, but the combination of * circular-arrays and supporting interior removes (i.e., those * not at head) would cause iterators to sometimes lose their * places and/or (re)report elements they shouldn't. To avoid * this, when a queue has one or more iterators, it keeps iterator * state consistent by: * * (1) keeping track of the number of "cycles", that is, the * number of times takeIndex has wrapped around to 0. * (2) notifying all iterators via the callback removedAt whenever * an interior element is removed (and thus other elements may * be shifted). * * These suffice to eliminate iterator inconsistencies, but * unfortunately add the secondary responsibility of maintaining * the list of iterators. We track all active iterators in a * simple linked list (accessed only when the queue's lock is * held) of weak references to Itr. The list is cleaned up using * 3 different mechanisms: * * (1) Whenever a new iterator is created, do some O(1) checking for * stale list elements. * * (2) Whenever takeIndex wraps around to 0, check for iterators * that have been unused for more than one wrap-around cycle. * * (3) Whenever the queue becomes empty, all iterators are notified * and this entire data structure is discarded. * * So in addition to the removedAt callback that is necessary for * correctness, iterators have the shutdown and takeIndexWrapped * callbacks that help remove stale iterators from the list. * * Whenever a list element is examined, it is expunged if either * the GC has determined that the iterator is discarded, or if the * iterator reports that it is "detached" (does not need any * further state updates). Overhead is maximal when takeIndex * never advances, iterators are discarded before they are * exhausted, and all removals are interior removes, in which case * all stale iterators are discovered by the GC. But even in this * case we don't increase the amortized complexity. * * Care must be taken to keep list sweeping methods from * reentrantly invoking another such method, causing subtle * corruption bugs. */
class Itrs {
/** * Node in a linked list of weak iterator references. */
private class Node extends WeakReference<Itr> {
Node next;
Node(Itr iterator, Node next) {
super(iterator);
this.next = next;
}
}
/** Incremented whenever takeIndex wraps around to 0 */
int cycles = 0;
/** Linked list of weak iterator references */
private Node head;
/** Used to expunge stale iterators */
private Node sweeper = null;
private static final int SHORT_SWEEP_PROBES = 4;
private static final int LONG_SWEEP_PROBES = 16;
Itrs(Itr initial) {
register(initial);
}
/** * Sweeps itrs, looking for and expunging stale iterators. * If at least one was found, tries harder to find more. * Called only from iterating thread. * * @param tryHarder whether to start in try-harder mode, because * there is known to be at least one iterator to collect */
void doSomeSweeping(boolean tryHarder) {
// assert lock.getHoldCount() == 1;
// assert head != null;
int probes = tryHarder ? LONG_SWEEP_PROBES : SHORT_SWEEP_PROBES;
Node o, p;
final Node sweeper = this.sweeper;
boolean passedGo; // to limit search to one full sweep
if (sweeper == null) {
o = null;
p = head;
passedGo = true;
} else {
o = sweeper;
p = o.next;
passedGo = false;
}
for (; probes > 0; probes--) {
if (p == null) {
if (passedGo)
break;
o = null;
p = head;
passedGo = true;
}
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.isDetached()) {
// found a discarded/exhausted iterator
probes = LONG_SWEEP_PROBES; // "try harder"
// unlink p
p.clear();
p.next = null;
if (o == null) {
head = next;
if (next == null) {
// We've run out of iterators to track; retire
itrs = null;
return;
}
}
else
o.next = next;
} else {
o = p;
}
p = next;
}
this.sweeper = (p == null) ? null : o;
}
/** * Adds a new iterator to the linked list of tracked iterators. */
void register(Itr itr) {
// assert lock.getHoldCount() == 1;
head = new Node(itr, head);
}
/** * Called whenever takeIndex wraps around to 0. * * Notifies all iterators, and expunges any that are now stale. */
void takeIndexWrapped() {
// assert lock.getHoldCount() == 1;
cycles++;
for (Node o = null, p = head; p != null;) {
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.takeIndexWrapped()) {
// unlink p
// assert it == null || it.isDetached();
p.clear();
p.next = null;
if (o == null)
head = next;
else
o.next = next;
} else {
o = p;
}
p = next;
}
if (head == null) // no more iterators to track
itrs = null;
}
/** * Called whenever an interior remove (not at takeIndex) occurred. * * Notifies all iterators, and expunges any that are now stale. */
void removedAt(int removedIndex) {
for (Node o = null, p = head; p != null;) {
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.removedAt(removedIndex)) {
// unlink p
// assert it == null || it.isDetached();
p.clear();
p.next = null;
if (o == null)
head = next;
else
o.next = next;
} else {
o = p;
}
p = next;
}
if (head == null) // no more iterators to track
itrs = null;
}
/** * Called whenever the queue becomes empty. * * Notifies all active iterators that the queue is empty, * clears all weak refs, and unlinks the itrs datastructure. */
void queueIsEmpty() {
// assert lock.getHoldCount() == 1;
for (Node p = head; p != null; p = p.next) {
Itr it = p.get();
if (it != null) {
p.clear();
it.shutdown();
}
}
head = null;
itrs = null;
}
/** * Called whenever an element has been dequeued (at takeIndex). */
void elementDequeued() {
// assert lock.getHoldCount() == 1;
if (count == 0)
queueIsEmpty();
else if (takeIndex == 0)
takeIndexWrapped();
}
}
/** * Iterator for ArrayBlockingQueue. * * To maintain weak consistency with respect to puts and takes, we * read ahead one slot, so as to not report hasNext true but then * not have an element to return. * * We switch into "detached" mode (allowing prompt unlinking from * itrs without help from the GC) when all indices are negative, or * when hasNext returns false for the first time. This allows the * iterator to track concurrent updates completely accurately, * except for the corner case of the user calling Iterator.remove() * after hasNext() returned false. Even in this case, we ensure * that we don't remove the wrong element by keeping track of the * expected element to remove, in lastItem. Yes, we may fail to * remove lastItem from the queue if it moved due to an interleaved * interior remove while in detached mode. */
private class Itr implements Iterator<E> {
/** Index to look for new nextItem; NONE at end */
private int cursor;
/** Element to be returned by next call to next(); null if none */
private E nextItem;
/** Index of nextItem; NONE if none, REMOVED if removed elsewhere */
private int nextIndex;
/** Last element returned; null if none or not detached. */
private E lastItem;
/** Index of lastItem, NONE if none, REMOVED if removed elsewhere */
private int lastRet;
/** Previous value of takeIndex, or DETACHED when detached */
private int prevTakeIndex;
/** Previous value of iters.cycles */
private int prevCycles;
/** Special index value indicating "not available" or "undefined" */
private static final int NONE = -1;
/** * Special index value indicating "removed elsewhere", that is, * removed by some operation other than a call to this.remove(). */
private static final int REMOVED = -2;
/** Special value for prevTakeIndex indicating "detached mode" */
private static final int DETACHED = -3;
Itr() {
// assert lock.getHoldCount() == 0;
lastRet = NONE;
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (count == 0) {
// assert itrs == null;
cursor = NONE;
nextIndex = NONE;
prevTakeIndex = DETACHED;
} else {
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
prevTakeIndex = takeIndex;
nextItem = itemAt(nextIndex = takeIndex);
cursor = incCursor(takeIndex);
if (itrs == null) {
itrs = new Itrs(this);
} else {
itrs.register(this); // in this order
itrs.doSomeSweeping(false);
}
prevCycles = itrs.cycles;
// assert takeIndex >= 0;
// assert prevTakeIndex == takeIndex;
// assert nextIndex >= 0;
// assert nextItem != null;
}
} finally {
lock.unlock();
}
}
boolean isDetached() {
// assert lock.getHoldCount() == 1;
return prevTakeIndex < 0;
}
private int incCursor(int index) {
// assert lock.getHoldCount() == 1;
if (++index == items.length)
index = 0;
if (index == putIndex)
index = NONE;
return index;
}
/** * Returns true if index is invalidated by the given number of * dequeues, starting from prevTakeIndex. */
private boolean invalidated(int index, int prevTakeIndex,
long dequeues, int length) {
if (index < 0)
return false;
int distance = index - prevTakeIndex;
if (distance < 0)
distance += length;
return dequeues > distance;
}
/** * Adjusts indices to incorporate all dequeues since the last * operation on this iterator. Call only from iterating thread. */
private void incorporateDequeues() {
// assert lock.getHoldCount() == 1;
// assert itrs != null;
// assert !isDetached();
// assert count > 0;
final int cycles = itrs.cycles;
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
final int prevCycles = this.prevCycles;
final int prevTakeIndex = this.prevTakeIndex;
if (cycles != prevCycles || takeIndex != prevTakeIndex) {
final int len = items.length;
// how far takeIndex has advanced since the previous
// operation of this iterator
long dequeues = (cycles - prevCycles) * len
+ (takeIndex - prevTakeIndex);
// Check indices for invalidation
if (invalidated(lastRet, prevTakeIndex, dequeues, len))
lastRet = REMOVED;
if (invalidated(nextIndex, prevTakeIndex, dequeues, len))
nextIndex = REMOVED;
if (invalidated(cursor, prevTakeIndex, dequeues, len))
cursor = takeIndex;
if (cursor < 0 && nextIndex < 0 && lastRet < 0)
detach();
else {
this.prevCycles = cycles;
this.prevTakeIndex = takeIndex;
}
}
}
/** * Called when itrs should stop tracking this iterator, either * because there are no more indices to update (cursor < 0 && * nextIndex < 0 && lastRet < 0) or as a special exception, when * lastRet >= 0, because hasNext() is about to return false for the * first time. Call only from iterating thread. */
private void detach() {
// Switch to detached mode
// assert lock.getHoldCount() == 1;
// assert cursor == NONE;
// assert nextIndex < 0;
// assert lastRet < 0 || nextItem == null;
// assert lastRet < 0 ^ lastItem != null;
if (prevTakeIndex >= 0) {
// assert itrs != null;
prevTakeIndex = DETACHED;
// try to unlink from itrs (but not too hard)
itrs.doSomeSweeping(true);
}
}
/** * For performance reasons, we would like not to acquire a lock in * hasNext in the common case. To allow for this, we only access * fields (i.e. nextItem) that are not modified by update operations * triggered by queue modifications. */
public boolean hasNext() {
// assert lock.getHoldCount() == 0;
if (nextItem != null)
return true;
noNext();
return false;
}
private void noNext() {
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
// assert cursor == NONE;
// assert nextIndex == NONE;
if (!isDetached()) {
// assert lastRet >= 0;
incorporateDequeues(); // might update lastRet
if (lastRet >= 0) {
lastItem = itemAt(lastRet);
// assert lastItem != null;
detach();
}
}
// assert isDetached();
// assert lastRet < 0 ^ lastItem != null;
} finally {
lock.unlock();
}
}
public E next() {
// assert lock.getHoldCount() == 0;
final E x = nextItem;
if (x == null)
throw new NoSuchElementException();
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (!isDetached())
incorporateDequeues();
// assert nextIndex != NONE;
// assert lastItem == null;
lastRet = nextIndex;
final int cursor = this.cursor;
if (cursor >= 0) {
nextItem = itemAt(nextIndex = cursor);
// assert nextItem != null;
this.cursor = incCursor(cursor);
} else {
nextIndex = NONE;
nextItem = null;
}
} finally {
lock.unlock();
}
return x;
}
public void remove() {
// assert lock.getHoldCount() == 0;
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (!isDetached())
incorporateDequeues(); // might update lastRet or detach
final int lastRet = this.lastRet;
this.lastRet = NONE;
if (lastRet >= 0) {
if (!isDetached())
removeAt(lastRet);
else {
final E lastItem = this.lastItem;
// assert lastItem != null;
this.lastItem = null;
if (itemAt(lastRet) == lastItem)
removeAt(lastRet);
}
} else if (lastRet == NONE)
throw new IllegalStateException();
// else lastRet == REMOVED and the last returned element was
// previously asynchronously removed via an operation other
// than this.remove(), so nothing to do.
if (cursor < 0 && nextIndex < 0)
detach();
} finally {
lock.unlock();
// assert lastRet == NONE;
// assert lastItem == null;
}
}
/** * Called to notify the iterator that the queue is empty, or that it * has fallen hopelessly behind, so that it should abandon any * further iteration, except possibly to return one more element * from next(), as promised by returning true from hasNext(). */
void shutdown() {
// assert lock.getHoldCount() == 1;
cursor = NONE;
if (nextIndex >= 0)
nextIndex = REMOVED;
if (lastRet >= 0) {
lastRet = REMOVED;
lastItem = null;
}
prevTakeIndex = DETACHED;
// Don't set nextItem to null because we must continue to be
// able to return it on next().
//
// Caller will unlink from itrs when convenient.
}
private int distance(int index, int prevTakeIndex, int length) {
int distance = index - prevTakeIndex;
if (distance < 0)
distance += length;
return distance;
}
/** * Called whenever an interior remove (not at takeIndex) occurred. * * @return true if this iterator should be unlinked from itrs */
boolean removedAt(int removedIndex) {
// assert lock.getHoldCount() == 1;
if (isDetached())
return true;
final int cycles = itrs.cycles;
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
final int prevCycles = this.prevCycles;
final int prevTakeIndex = this.prevTakeIndex;
final int len = items.length;
int cycleDiff = cycles - prevCycles;
if (removedIndex < takeIndex)
cycleDiff++;
final int removedDistance =
(cycleDiff * len) + (removedIndex - prevTakeIndex);
// assert removedDistance >= 0;
int cursor = this.cursor;
if (cursor >= 0) {
int x = distance(cursor, prevTakeIndex, len);
if (x == removedDistance) {
if (cursor == putIndex)
this.cursor = cursor = NONE;
}
else if (x > removedDistance) {
// assert cursor != prevTakeIndex;
this.cursor = cursor = dec(cursor);
}
}
int lastRet = this.lastRet;
if (lastRet >= 0) {
int x = distance(lastRet, prevTakeIndex, len);
if (x == removedDistance)
this.lastRet = lastRet = REMOVED;
else if (x > removedDistance)
this.lastRet = lastRet = dec(lastRet);
}
int nextIndex = this.nextIndex;
if (nextIndex >= 0) {
int x = distance(nextIndex, prevTakeIndex, len);
if (x == removedDistance)
this.nextIndex = nextIndex = REMOVED;
else if (x > removedDistance)
this.nextIndex = nextIndex = dec(nextIndex);
}
else if (cursor < 0 && nextIndex < 0 && lastRet < 0) {
this.prevTakeIndex = DETACHED;
return true;
}
return false;
}
/** * Called whenever takeIndex wraps around to zero. * * @return true if this iterator should be unlinked from itrs */
boolean takeIndexWrapped() {
// assert lock.getHoldCount() == 1;
if (isDetached())
return true;
if (itrs.cycles - prevCycles > 1) {
// All the elements that existed at the time of the last
// operation are gone, so abandon further iteration.
shutdown();
return true;
}
return false;
}
// /** Uncomment for debugging. */
// public String toString() {
// return ("cursor=" + cursor + " " +
// "nextIndex=" + nextIndex + " " +
// "lastRet=" + lastRet + " " +
// "nextItem=" + nextItem + " " +
// "lastItem=" + lastItem + " " +
// "prevCycles=" + prevCycles + " " +
// "prevTakeIndex=" + prevTakeIndex + " " +
// "size()=" + size() + " " +
// "remainingCapacity()=" + remainingCapacity());
// }
}
/** * Returns a {@link Spliterator} over the elements in this queue. * * <p>The returned spliterator is * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>. * * <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT}, * {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}. * * @implNote * The {@code Spliterator} implements {@code trySplit} to permit limited * parallelism. * * @return a {@code Spliterator} over the elements in this queue * @since 1.8 */
public Spliterator<E> spliterator() {
return Spliterators.spliterator
(this, Spliterator.ORDERED | Spliterator.NONNULL |
Spliterator.CONCURRENT);
}
}
