基于 JDK 1.8.0
1. 概括
ArrayBlockingQueue 是一个先进先出(FIFO)的阻塞队列,底层是数组,队列长度在创建的时候确定不能修改。
使用场景:生产者消费者
2. 类定义
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable
ArrayBlockingQueue 继承于 AbstractQueue,实现 BlockingQueue,Serializable 接口。
BlockingQueue 的介绍参考 BlockingQueue 源码学习
AbstractQueue 源码查看
public abstract class AbstractQueue<E>
extends AbstractCollection<E>
implements Queue<E>
抽象类 AbstractQueue 继承于 AbstractCollection,实现 Queue 接口。
其中 AbstractQueue 实现了 Queue 的一些操作,在插入时不允许为 NULL;
根据 AbstractQueue 源码知道,AbstractQueue 对 Queue 接口的方法的返回值进行的参数值的判断,如果添加失败或者没有查找到元素,就会抛出相关异常,这里就不再展示。
3. 成员变量
/**
* 队列中的元素,final 类型不可变
*/
final Object[] items;
/** 下一次执行 take,poll,peek,remove 操作的位置 */
int takeIndex;
/** 下一次执行 put,offer,add 操作的位置,在哪个位置进行这些操作 */
int putIndex;
/** 队列中的元素个数 */
int count;
/** 锁,所有请求都会先去获取这个锁 */
final ReentrantLock lock;
/** 条件队列,用于通知 take 方法告知队列中有元素了 */
private final Condition notEmpty;
/** 条件队列,用于通知 put() 方法有空间放元素了 */
private final Condition notFull;
/**
* TODO 迭代器,后面再展开学习
* Shared state for currently active iterators, or null if there
* are known not to be any. Allows queue operations to update
* iterator state.
*/
transient Itrs itrs = null;
根据成员变量的介绍可以得知:
- 该队列用数组保持元素
- 入队和出队用了两个指针标记,这两个标记的移动类似于在环形节点上面移动,如上图,下面详细分析。
- 并发控制使用了 ReentrantLock ,并且使用两个 Condition 来控制队列的入队方法和出队方法的阻塞与继续运行
4. 构造方法
/**
* 仅指定容量大小,默认使用 非公平锁 控制并发请求
*/
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
/**
* 生成指定容量大小的数组,容量必须大于 0
* 构造 锁 和 条件队列
*/
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 中的元素添加到 ArrayBlockingQueue 中
* count = c.size()
* 若集合 c 的容量大小等于 capacity,将 putIndex 置为 0
*/
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. 成员方法
5.1 入队方法
5.1.1 boolean add(E e)
调用父类 add 方法,而父类 AbstractQueue add 方法实际调用的是 offer(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");
}
5.1.2 boolean offer(E e) 与 boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException
- 加锁后判断容量大小,满了就不添加;没满执行共用的入队方法,入队成功返回 true
/**
* 插入成功返回 true,否则返回 false
* 插入元素为 NULL 返回空指针异常
*/
public boolean offer(E e) {
checkNotNull(e); // 首先判断元素是否为 NULL,为 NULL 抛出异常
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false; // 若数组满,返回 false
else {
enqueue(e); // 添加元素并返回 true
return true;
}
} finally {
lock.unlock();
}
}
/**
* 带超时时间的添加元素方法
* 当超时时间达到,直接返回 false
*/
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();
}
}
内部共用的 入队 方法 enqueue()
- 仅在同步方法块中调用
- putIndex 的值一直在从 0 到 items.length 之间循环,添加元素进来就往后移动一位,移动到最后就又回到了开头,像是在环形数组上面移动
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal(); // 通知等待 take 的线程,表示当前数组有元素可取
}
5.1.3 void put(E e) 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();
}
}
入队方法总结:
- 根据 BlockingQueue 源码可知,添加元素有 3 个方法,分别是 add,offer,put,其中 add 方法添加元素失败会抛出异常,官方推荐使用 offer,put 方法带阻塞功能。
- enqueue 是内部共用的添加元素方法,其中 putIndex 的值一直都在从表头 -> 表尾 -> 表头循环,有点类似在环上面移动
- ArrayBlockingQueue 的 add,offer,put 方法都是线程安全的
5.2 出队方法
5.2.1 E poll() 与 E poll(long timeout, TimeUnit unit) throws InterruptedException
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue(); // 若队列没有元素,返回 NULL;否则执行出队操作,将表头节点弹出并返回
} 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; // 等待超时,返回 NULL
nanos = notEmpty.awaitNanos(nanos);
}
return dequeue();
} finally {
lock.unlock();
}
}
dequeue 内部共用的出队操作
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 往后移动一位,若 takeIndex 已经到数组末尾,那么让他回到开头
takeIndex = 0;
count--; // 队列数组元素减 1
if (itrs != null)
itrs.elementDequeued();
notFull.signal(); // 通知等待 notFull 变量可用的线程,队列将可以添加元素进来
return x;
}
5.2.2 E take() throws InterruptedException
- 与 poll 不同的是,take 取元素会阻塞
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await(); // 没有元素时进行等待
return dequeue(); // 否则取出表头节点
} finally {
lock.unlock();
}
}
5.3 获取表头元素
- 直接取 takeIndex 处的元素,队列为空时返回 NULL
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
5.4 删除指定元素
- 首先这个方法一定是在同步方法块中进行移除操作 ,环形图继续拿出来,假设原来是顺时针方向进行插入元素
- 删除元素,就得从 takeIndex (表头位置)开始遍历,如果有相同的就移除当前位置的元素
- 若 takeIndex 走到末尾,然后再置为 0,就像上面环一样,一直顺时针走
- 若 takeIndex = putIndex ,说明走到遍历到末尾了,注意这里的末尾是插入位置 putIndex 的前一个位置,说明前面的位置已经遍历完了。
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 到表头
i = 0;
} while (i != putIndex); // i == putIndex 时,说明数组中元素已经全部遍历,因为 putIndex 是下一次插入元素的位置
}
return false; // 队列为空,删除不了元素,返回 false
} finally {
lock.unlock();
}
}
/**
* 仅在获取到锁时调用,删除的过程其实就是 逆时针 移动元素的过程
*/
void removeAt(final int removeIndex) {
// assert lock.getHoldCount() == 1;
// assert items[removeIndex] != null;
// assert removeIndex >= 0 && removeIndex < items.length;
final Object[] items = this.items;
// 若待删除元素位置等于此时表头位置,直接将此处元素置为 NULL,并且 takeIndex 指向下一个位置,元素无需移动(参考上面环形图,移除 takeIndex 元素只需要置 NULL 然后后移)
if (removeIndex == takeIndex) {
// 元素值直接置为 null
items[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;
// 待删除位置已经是最后一个元素,将 next 置为 0
if (next == items.length)
next = 0;
if (next != putIndex) {
// 逆时针移动,把后面的元素移到前面来
items[i] = items[next];
i = next;
} else {
// 此处位置的下一个位置是 putIndex,那么把此处位置置为 null
items[i] = null;
this.putIndex = i;
break;
}
}
// 更新数组元素数量
count--;
if (itrs != null)
// 更新迭代器
itrs.removedAt(removeIndex);
}
notFull.signal(); // 唤醒等待的 put 线程
}
void elementDequeued() {
// assert lock.getHoldCount() == 1;
if (count == 0)
queueIsEmpty();
else if (takeIndex == 0)
takeIndexWrapped();
}
内部类( TODO 后面再补充)
class Itrs {
}
private class Itr implements Iterator<E> {
}