面试题
- ArrayBlockingQueue如何实现并发安全的
- ArrayBlockingQueue如何实现阻塞的
不想看源码解析的同学，可以直接去最下方查看答案
源码解析
ArrayBlockingQueue是底层由数组实现的有界阻塞队列。ArrayBlockingQueue满足队列的FIFO(先入先出)特性。
ArrayBlockingQueue长度是固定的，创建后，长度就无法修改了。
ArrayBlockingQueue默认情况是非公平的，也就是说当多个 生产者/消费者 尝试向队列中添加/移除元素时，顺序是不固定的。可以在调用构造函数时，通过设置fair为true，改为公平队列。
底层结构
/**
* 保存元素的数组
*/
final Object[] items;
/**
* 下一次take，poll，peek或者remove操作返回的数据下标
*/
int takeIndex;
/**
* 下一次put，offer或者add要操作的下标
*/
int putIndex;
/**
* queue中元素的数量
*/
int count;
/*
* 双条件算法控制并发安全
*/
/**
* Main lock guarding all access
*/
final ReentrantLock lock;
/**
* take的等待条件(当take时如果队列为空则会在此条件上等待)
*/
private final Condition notEmpty;
/**
* put的等待条件(当put时如果队列已满则会在此条件上等待)
*/
private final Condition notFull;
/**
* 当前活跃(生效)的迭代器的共享状态，如果没有活跃迭代器，则为null
* 允许queue操作更新迭代器状态
*/
transient Itrs itrs = null;
内部工具方法
/**
* 循环递减i，因为元素在数组中是按环状保存的
*/
final int dec(int i) {
return ((i == 0) ? items.length : i) - 1;
}
/**
* 返回数组中下标i对应的元素
*/
@SuppressWarnings("unchecked")
final E itemAt(int i) {
return (E) items[i];
}
// ArrayBlockingQueue不允许添加null元素，所以当添加的
// 元素为null时，会抛出NullPointerException
private static void checkNotNull(Object v) {
if (v == null)
throw new NullPointerException();
}
/**
* 在当前位置(队列尾)插入元素，移动下标，释放notEmpty条件
*/
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
// putIndex位置对应的是队列尾部，在尾部插入元素
items[putIndex] = x;
// 循环递增下标
if (++putIndex == items.length)
putIndex = 0;
count++;
// 唤醒阻塞在notEmpty条件的消费者线程
notEmpty.signal();
}
/**
* 移除当前位置的元素(队列头)，移动下标，释放notFull条件
*/
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
// takeIndex位置对应的是队列头部，移除队列头部元素
items[takeIndex] = null;
// 循环递增下标
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
// 通知所有迭代器
itrs.elementDequeued();
// 唤醒阻塞在notFull条件的生产者线程
notFull.signal();
return x;
}
/**
* 删除removeIndex对应的元素
*/
void removeAt(final int removeIndex) {
// assert lock.getHoldCount() == 1;
// assert items[removeIndex] != null;
// assert removeIndex >= 0 && removeIndex < items.length;
final Object[] items = this.items;
if (removeIndex == takeIndex) {
// 如果移除元素的下表恰好是队列头部，就按普通出队的方式
// 处理就可以
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
} else {
// an "interior" remove
// putIndex对应的是下一个要添加元素的位置(队列尾)
final int putIndex = this.putIndex;
// 从removeIndex遍历到队列尾，将removeIndex之后的元素全部向前移动
for (int i = removeIndex; ; ) {
// 循环处理下一个元素下标
int next = i + 1;
if (next == items.length)
next = 0;
if (next != putIndex) {
// 如果下一个元素不是队列尾(队列尾是下一个要入队的
// 位置，不存在元素)，则将元素向前移动一个位置
items[i] = items[next];
// 下标后移
i = next;
} else {
// 如果next对应的是下一个入队的位置，则i对应的是最后一个元素
// 因为上一次循环中已经向前移动过了，这里直接删除多余元素,
// 并重置队列尾为i
items[i] = null;
this.putIndex = i;
break;
}
}
count--;
// 调用迭代器进行删除
if (itrs != null)
itrs.removedAt(removeIndex);
}
// 唤醒阻塞在notFull条件的生产者线程
notFull.signal();
}
构造方法
/**
* 通过给定容量构造非公平阻塞队列，初始化后容量无法修改
*/
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();
}
/**
* 通过给定容量和是否公平参数 构造阻塞队列，初始化后容量无法修改。
* 并将给定集合c中的元素添加到队列中
*/
public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) {
this(capacity, fair);
// 插入操作前需要先锁住整个队列
final ReentrantLock lock = this.lock;
// 因为是构造方法，这里锁是为了保证可见性而不是互斥性
lock.lock();
try {
int i = 0;
try {
// 遍历集合c，并将元素添加到阻塞队列中
for (E e : c) {
checkNotNull(e);
// 这里并没有对下标i进行处理，所以如果c的容量大于capacity时
// 会抛出ArrayIndexOutOfBoundsException
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;
// 循环处理下一次要入队的下标(队列尾)
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock();
}
}
插入
/**
* 插入给定元素e到队列尾部。如果插入时队列已满，则抛出IllegalStateException
*/
public boolean add(E e) {
// 内部调用offer方法
// if (offer(e))
// return true;
// else
// throw new IllegalStateException("Queue full");
return super.add(e);
}
/**
* 插入给定元素e到队列尾部。如果插入时队列已满，则返回false
*/
public boolean offer(E e) {
// ArrayBlockingQueue不允许元素为null
checkNotNull(e);
// 插入操作前必须要先锁住队列
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
// 如果队列已满，返回false
return false;
else {
// 入队
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}
/**
* 插入给定元素e到队列尾部。如果插入时队列已满，则阻塞直到队列有空间
*/
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
// 如果队列已满则在notFull条件等待，直到其他消费者
// 从队列中删除元素后调用notFull.signal()唤醒线程
notFull.await();
// 入队
enqueue(e);
} finally {
lock.unlock();
}
}
/**
* 插入给定元素e到队列尾部。如果插入时队列已满，则阻塞直到
* 1. 队列有空间
* 2. 超时退出
*/
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;
// 在notFull条件等待，等待时间为剩余的timeout时间
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
删除/查询
/**
* 删除队列头并返回，如果队列为空则返回null
*/
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
/**
* 删除队列头并返回，如果队列为空，则阻塞直到有其他线程
* 向队列内添加元素，释放notEmpty条件
*/
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
// 如果队列为空，则在notEmpty条件等待，直到
// 有其他线程插入元素后调用notEmpty.signal()
// 唤醒该线程
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
/**
* 删除队列头并返回，如果队列为空，则阻塞直到
* 1. 其他线程向队列内插入元素
* 2. 超时返回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();
}
}
/**
* 返回队列头但不删除，如果队列为空则返回null
*/
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
/**
* 如果队列中存在一个或多个对象o，则删除队列中最早的一个o，并返回true。
* 否则返回false
*/
public boolean remove(Object o) {
// 因为队列中不允许值为null，所以o为null直接返回false
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count > 0) {
// putIndex为下一个要插入元素的位置(队尾)
final int putIndex = this.putIndex;
// takeIndex为队列头
int i = takeIndex;
// 遍历找到第一个o，并删除
do {
if (o.equals(items[i])) {
removeAt(i);
return true;
}
// 因为数组是循环利用的，所以需要按循环处理
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
// 如果队列为空，直接返回false
return false;
} finally {
lock.unlock();
}
}
其他
/**
* 返回队列内元素数量
*/
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
/**
* 返回队列中剩余容量
*/
public int remainingCapacity() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return items.length - count;
} finally {
lock.unlock();
}
}
/**
* 如果队列中包含元素o，则返回true
*/
public boolean contains(Object o) {
// 因为队列中不允许值为null，所以o为null直接返回false
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count > 0) {
// putIndex为下一个要插入元素的位置(队尾)
final int putIndex = this.putIndex;
// takeIndex为队列头
int i = takeIndex;
// 遍历队列
do {
if (o.equals(items[i]))
// 如果相等直接返回true
return true;
// 因为数组是循环利用的，所以需要按循环处理
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
// 如果队列为空直接返回false
return false;
} finally {
lock.unlock();
}
}
迭代器
/**
* 迭代器及其队列之间的共享数据，允许在删除元素时更新迭代器
* <p>
* 因为继承了Collection的一些特性，为了支持从非头部删除元素等
* 操作造成的迭代错误。
* <p>
* 当队列具有一个或多个迭代器时，它通过以下方式使迭代器状态保持一致：
* 1. 追踪cycles字段
* 2. 每当删除内部元素（并因此可能移动其他元素）时，都通过回调removeAt
* 通知所有迭代器
* <p>
* 这些足以消除迭代器的不一致，但是需要维护iterators。我们在对Itr的弱引
* 用的链表中跟踪所有活动的迭代器。该列表使用3种不同的机制进行清理：
* <p>
* 1. 每当创建新的迭代器时，都要检查过时的迭代器
* 2. 每当takeIndex绕回0时，需要检查超过一回合没有使用过的Iterator
* 3. 每当queue变为空的时候，通知所有迭代器，并丢弃整个数据结构。
* <p>
* 每当检查list中的迭代器时，如果GC确定丢弃了迭代器，或者如果迭代器
* 报告它是detached(不再需要任何更新)则将其删除。
*/
class Itrs {
/**
* 保存了每个活跃迭代器的弱引用
*/
private class Node extends WeakReference<Itr> {
Node next;
Node(Itr iterator, Node next) {
super(iterator);
this.next = next;
}
}
/**
* 每当takeIndex变成0时递增，代表itrs跟踪的最新循环(相当于版本号)
*/
int cycles = 0;
/**
* 链表头结点
*/
private Node head;
/**
* 用于删除过时的迭代器
*/
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);
}
/**
* 扫描维护的迭代器列表，寻找并删除失效的迭代器。
* <p>
* 如果找到至少一个，则尝试扫描更多。仅从迭代线程调用。
*
* @param tryHarder 是否以try-harder模式启动，true-已知至少有一个迭代器要收集
*/
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; // 将搜索限制为一次完整扫描
if (sweeper == null) {
// 如果上一个遍历过的结点为null(说明没有遍历过，或者
// 上一次全部遍历了一次)，则从头开始遍历
o = null;
p = head;
passedGo = true;
} else {
// p为下一个需要遍历的迭代器
o = sweeper;
p = o.next;
passedGo = false;
}
for (; probes > 0; probes--) {
if (p == null) {
// 如果p为null，说明没有需要扫描的迭代器了
if (passedGo)
// 如果限制了一次扫描，则直接退出
break;
// 设置上一个扫描的迭代器为null
o = null;
// 下一个从head迭代器开始扫描
p = head;
passedGo = true;
}
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.isDetached()) {
// p对应的迭代器it是 废弃/失效 的迭代器
// 使用try-harder模式，继续扫描
probes = LONG_SWEEP_PROBES; // "try harder"
// 移除p的引用
p.clear();
p.next = null;
// 从迭代器链表移除p结点
if (o == null) {
head = next;
if (next == null) {
// 所有维护的迭代器都无效了，将链表置为null
itrs = null;
return;
}
} else
o.next = next;
} else {
// 迭代器it没有失效，向后遍历
o = p;
}
p = next;
}
// p == null 说明没有需要遍历的迭代器了，返回null，否则返回上一个遍历过的迭代器o
this.sweeper = (p == null) ? null : o;
}
/**
* 将新的迭代器添加到跟踪的迭代器链表中
*/
void register(Itr itr) {
// assert lock.getHoldCount() == 1;
head = new Node(itr, head);
}
/**
* 当takeIndex重新变回0时，调用该方法
* <p>
* 通知所有迭代器，并清除所有过时的迭代器。
*/
void takeIndexWrapped() {
// assert lock.getHoldCount() == 1;
// 循环+1
cycles++;
// 遍历所有维护的迭代器
for (Node o = null, p = head; p != null; ) {
final Itr it = p.get();
final Node next = p.next;
// 如果当前迭代器为引用为null，或者循环导致迭代器失效
// 从维护的迭代器列表中移除该迭代器
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;
}
/**
* 当从非队列头的其他位置删除元素时，会调用该方法
* <p>
* 通知所有迭代器，并清除所有过时的迭代器。
*/
void removedAt(int removedIndex) {
for (Node o = null, p = head; p != null; ) {
final Itr it = p.get();
final Node next = p.next;
// 如果当前迭代器为引用为null，或者删除导致迭代器失效
// 从维护的迭代器列表中移除该迭代器
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;
}
/**
* 当删除元素导致队列为空时调用该方法
* <p>
* 通知所有活跃的迭代器，queue已经为空，清空所有weak引用，
* 并且unlink itrs数据结构
*/
void queueIsEmpty() {
// assert lock.getHoldCount() == 1;
for (Node p = head; p != null; p = p.next) {
Itr it = p.get();
if (it != null) {
// 清空weak引用
p.clear();
// 通知迭代器queue已经为空
it.shutdown();
}
}
head = null;
itrs = null;
}
/**
* 当有元素被删除时，调用该方法
*/
void elementDequeued() {
// assert lock.getHoldCount() == 1;
if (count == 0)
// 通知所有活跃迭代器，队列已经为空
queueIsEmpty();
else if (takeIndex == 0)
// 通知所有活跃迭代器，并移除过时迭代器
takeIndexWrapped();
}
}
/**
* 当所有索引均为负数时，或者当hasNext首次返回false时，将切换到
* "DETACHED"模式（允许在不使用GC的情况下迅速从itr取消链接）。
* 这使迭代器可以准确地跟踪并发更新(但在hasNext()返回false之后
* 用户调用Iterator.remove()的特殊情况除外)。
* <p>
* 即使在这种情况下，也可以确保通过在lastItem中跟踪要删除的预期
* 元素来确保不会删除错误的元素。
* <p>
* 但是由于在DETACHED模式下交错插入的内部移除，如果lastItem移
* 动了，可能无法从队列中移除它。
*/
private class Itr implements Iterator<E> {
/**
* 下一个要遍历的元素，如果是队尾则返回NONE
*/
private int cursor;
/**
* next()方法返回的元素，如果没有则返回null
*/
private E nextItem;
/**
* nextItem对应的下标，如果没有则返回NONE，如果被移除了就返回REMOVED
*/
private int nextIndex;
/**
* 最后返回的元素；如果没有或在未detach状态，则为null。
*/
private E lastItem;
/**
* lastItem对应的下标，如果没有则返回NONE，如果被移除了就返回REMOVED
*/
private int lastRet;
/**
* 上次遍历操作时的队列头，或者在detach状态下为DETACHED
*/
private int prevTakeIndex;
/**
* iters.cycles的上一个值
*/
private int prevCycles;
/**
* 表示"not available" or "undefined"(不可用或未定义)
*/
private static final int NONE = -1;
/**
* 表示"removed elsewhere"(已被其他线程删除)
*/
private static final int REMOVED = -2;
/**
* 表示"detached mode"(DETACHED状态)
*/
private static final int DETACHED = -3;
Itr() {
// assert lock.getHoldCount() == 0;
// 构造迭代器，上一个返回的元素下标为NONE
lastRet = NONE;
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (count == 0) {
// 如果队列为空
// assert itrs == null;
// 没有元素需要访问，下一个要访问的数据下标为NONE
cursor = NONE;
nextIndex = NONE;
prevTakeIndex = DETACHED;
} else {
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
// 下一个要访问的即为队列头
prevTakeIndex = takeIndex;
nextItem = itemAt(nextIndex = takeIndex);
// 返回takeIndex对应的下一个元素的下标，或者全部遍
// 历完则返回NONE
cursor = incCursor(takeIndex);
if (itrs == null) {
// 如果itrs为null，则初始化
itrs = new Itrs(this);
} else {
// 否则注册当前迭代器到itrs中
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();
}
}
// 判断是否为DETACHED模式
boolean isDetached() {
// assert lock.getHoldCount() == 1;
return prevTakeIndex < 0;
}
// 向后移动一位下标
private int incCursor(int index) {
// assert lock.getHoldCount() == 1;
// 处理循环移动的情况
if (++index == items.length)
index = 0;
// 如果下标已经移动到队列尾了，就说明遍历完
// 所有元素了，返回NONE
if (index == putIndex)
index = NONE;
return index;
}
/**
* 检查index是否有效
*
* @param index 需要检查的下标
* @param prevTakeIndex 上次遍历时对应的队列头
* @param dequeues 两次遍历间元素出队的次数
* @param length 现在队列中元素的数量
*/
private boolean invalidated(int index, int prevTakeIndex,
long dequeues, int length) {
if (index < 0)
return false;
// index与上一次队列头之间的距离
int distance = index - prevTakeIndex;
// 如果distance小于0，说明经过了一个循环，需要
// 加上队列中元素的数量
if (distance < 0)
distance += length;
// 检查队列头移动的距离和index移动的距离，判断是否有效
return dequeues > distance;
}
/**
* 调整索引，合并自此迭代器上一次操作以来的所有出队操作。仅从迭代线程调用。
*/
private void incorporateDequeues() {
// assert lock.getHoldCount() == 1;
// assert itrs != null;
// assert !isDetached();
// assert count > 0;
// 当前itrs维护的最新循环
final int cycles = itrs.cycles;
// 当前队列头
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
// 当前迭代器所属的循环
final int prevCycles = this.prevCycles;
// 上一次取到的下标
final int prevTakeIndex = this.prevTakeIndex;
// 如果当前迭代器的循环和itrs维护的最新循环不一致，或者
// 上次遍历时的队列头与现在队列头不一致。
// 说明两次遍历操作之间有出队操作
if (cycles != prevCycles || takeIndex != prevTakeIndex) {
// 当前队列中元素个数
final int len = items.length;
// 自该迭代器的上一次操作以来，takeIndex移动了多少
long dequeues = (cycles - prevCycles) * len
+ (takeIndex - prevTakeIndex);
// 检查 lastRet，nextIndex，cursor是否有效
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模式
detach();
else {
// 更新循环和队列头
this.prevCycles = cycles;
this.prevTakeIndex = takeIndex;
}
}
}
/**
* 当itrs需要停止维护此迭代器时调用此方法。以下几种原因可能需要停止维护：
* 1. 没有更多元素需要更新 cursor < 0 && nextIndex < 0 && lastRet < 0
* 2. 当lastRet >= 0时，因为hasNext()第一次返回false
*/
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;
// 将迭代器设置为DETACHED模式
prevTakeIndex = DETACHED;
// 扫描维护的迭代器列表，寻找并删除失效的迭代器。
itrs.doSomeSweeping(true);
}
}
/**
* 出于性能考虑，在通常情况下，调用hasNext时不上锁。因此，仅访问nextItem
* 字段，因为队列remove触发的更新操作不会修改nextItem
*/
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();
if (lastRet >= 0) {
// 如果lastRet有效
lastItem = itemAt(lastRet);
// assert lastItem != null;
// 没有元素需要迭代了，将迭代器设置为DETACHED状态
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) {
// 下一次next要访问的元素
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();
// 设置上一个元素的下标无效，因为要删除了
final int lastRet = this.lastRet;
this.lastRet = NONE;
if (lastRet >= 0) {
if (!isDetached())
removeAt(lastRet);
else {
// DETACH模式lastItem，即为迭代器代表的最后一个元素
final E lastItem = this.lastItem;
// assert lastItem != null;
this.lastItem = null;
// 如果lastRet没被删除过，则删除
if (itemAt(lastRet) == lastItem)
removeAt(lastRet);
}
} else if (lastRet == NONE)
throw new IllegalStateException();
if (cursor < 0 && nextIndex < 0)
detach();
} finally {
lock.unlock();
// assert lastRet == NONE;
// assert lastItem == null;
}
}
/**
* 调用此方法以通知迭代器该队列为空，或已落后于队列。因此它应放弃任何进
* 一步的迭代。
* <p>
* 但有一种可能hasNext已经返回true了，下一次调用next()必须返回元素。
*/
void shutdown() {
// assert lock.getHoldCount() == 1;
// 放弃迭代，下一个要遍历的元素为NONE
cursor = NONE;
// 由于迭代器已经过时，所以构造迭代器时期望要遍历的元素
// 都已过期(被移除)。将index置为REMOVED
if (nextIndex >= 0)
nextIndex = REMOVED;
if (lastRet >= 0) {
lastRet = REMOVED;
lastItem = null;
}
prevTakeIndex = DETACHED;
// 不能将nextItem置为null，因为可能next()要返回
}
private int distance(int index, int prevTakeIndex, int length) {
int distance = index - prevTakeIndex;
if (distance < 0)
distance += length;
return distance;
}
/**
* 该方法只需要处理由于删除元素导致的各个index的变化，不需要负责
* 删除数据，在其他方法中处理过了
*
* @return true-如果迭代器不需要被维护了
*/
boolean removedAt(int removedIndex) {
// assert lock.getHoldCount() == 1;
if (isDetached())
return true;
// itrs维护的最新循环(版本)
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++;
// 计算removedIndex与prevTakeIndex距离
final int removedDistance =
(cycleDiff * len) + (removedIndex - prevTakeIndex);
// assert removedDistance >= 0;
int cursor = this.cursor;
if (cursor >= 0) {
// cursor与prevTakeIndex距离
int x = distance(cursor, prevTakeIndex, len);
if (x == removedDistance) {
// 如果要删除的元素是下一个要访问的元素
if (cursor == putIndex)
// 如果下一个访问的位置是队尾，cursor无效
this.cursor = cursor = NONE;
} else if (x > removedDistance) {
// 要删除的元素在队列中，因为要删除元素cursor递减
// 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)
// lastRet递减
this.lastRet = lastRet = dec(lastRet);
}
int nextIndex = this.nextIndex;
if (nextIndex >= 0) {
int x = distance(nextIndex, prevTakeIndex, len);
if (x == removedDistance)
// 如果要删除的元素是下一次next()应该返回的元素
// 设置为已移除
this.nextIndex = nextIndex = REMOVED;
else if (x > removedDistance)
this.nextIndex = nextIndex = dec(nextIndex);
} else if (cursor < 0 && nextIndex < 0 && lastRet < 0) {
// 设置DETACH状态
this.prevTakeIndex = DETACHED;
return true;
}
return false;
}
/**
* 每当takeIndex变为0时，调用该方法
*
* @return 如果该迭代器应该被从itrs中移除则返回true
*/
boolean takeIndexWrapped() {
// assert lock.getHoldCount() == 1;
if (isDetached())
// DETACHED模式直接返回true
return true;
if (itrs.cycles - prevCycles > 1) {
// 因为只有takeIndex变为0时，itrs.cycles才会递增
// 而takeIndex可能变为0的原因就是出队，所以该条件
// 表示距离上一次操作时存在的所有元素均已出队，因
// 此放弃进一步的迭代。
shutdown();
return true;
}
return false;
}
}
面试题解答
-
ArrayBlockingQueue如何实现并发安全的
ArrayBlockingQueue底层使用了ReentrantLock，并在所有的put，take，size等操作前先获取这个锁。只有获取锁成功后，才允许对BlockingQueue进行操作。
-
ArrayBlockingQueue如何实现阻塞的
ArrayBlockingQueue在底层使用的lock上新建了两个条件变量，notEmpty和notFull
- 当生产者执行put等操作时，首先查看队列是否已满，如果已满会调用
notFull.await阻塞等待。put成功时会调用notEmpty.signal()唤醒等待的消费者线程。 - 当消费者执行take等操作时，首先查看队列是否为空，如果为空会调用
notEmpty.await阻塞等待。take成功时会调用notFull.signal()唤醒等待的生产者线程。
- 当生产者执行put等操作时，首先查看队列是否已满，如果已满会调用
