开篇
ArrayBlockingQueue是数组实现的线程安全的有界的阻塞队列。
- 线程安全是指ArrayBlockingQueue内部通过“互斥锁”保护竞争资源，实现了多线程对竞争资源的互斥访问。
- 有界是指ArrayBlockingQueue对应的数组是有界限的。
- 阻塞队列是指多线程访问竞争资源时，当竞争资源已被某线程获取时，其它要获取该资源的线程需要阻塞等待；而且，ArrayBlockingQueue是按 FIFO（先进先出）原则对元素进行排序，元素都是从尾部插入到队列，从头部开始返回。
ArrayBlockingQueue类图
ArrayBlockingQueue类图
ArrayBlockingQueue构造函数
ArrayBlockingQueue的构造函数信息表明以下几个信息：
- 线程安全 ReentrantLock lock
- 容量有界 this.items = new Object[capacity];
- 状态同步 Condition notEmpty、Condition notFull
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
final Object[] items;
int takeIndex;
int putIndex;
int count;
// 通过lock来保证线程安全，通过lock下的Condition来实现状态同步
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
transient Itrs itrs = null;
// 构造函数必须指定数组大小，所以是有界的
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 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();
}
}
}
ArrayBlockingQueue常用操作
ArrayBlockingQueue的add/put/offer方法
ArrayBlockingQueue的所有add()方法其实执行的就是offer()方法，其他核心的逻辑在enqueue方法中。当然所有操作都是执行加锁操作lock.lock()。
- add/offer方法是非阻塞的，如果队列满就直接返回异常
- put()方法是阻塞的，如果队列满就等待，等待notFull的信号量，notFull.await()在take等方法执行的时候会触发notFull.signal()。
- enqueue()方法内部就是往item数组中添加元素、计算元素个数count++、重置putIndex、通知非空信号量notEmpty.signal()。
public boolean add(E e) {
return super.add(e);
}
// 将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量），在成功时返回 true，如果此队列已满，则抛出 IllegalStateException。
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();
}
}
// 将指定的元素插入此队列的尾部，如果该队列已满，则等待可用的空间。
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();
}
}
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();
}
ArrayBlockingQueue的poll/take/peek方法
ArrayBlockingQueue的所有take相关操作最终都是执行dequeue操作的。
- 所有删除元素操作都是先进行加锁保证线程安全
- poll()和take()方法是阻塞的
- take()和poll()带时间参数是阻塞
- dequeue()内部通过takeIndex参数获取待返回的参数，重置元素个数count，移动下一个元素位置takeIndex等。
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : 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();
}
}
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 peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}
ArrayBlockingQueue的内部的锁
在ArrayBlockingQueue函数内部的加锁动作，我们发现有lock和lockInterruptibly两种，lock 与 lockInterruptibly比较区别在于：
- lock 优先考虑获取锁，待获取锁成功后，才响应中断。
- lockInterruptibly 优先考虑响应中断，而不是响应锁的普通获取或重入获取。
详细区别：
- ReentrantLock.lockInterruptibly允许在等待时由其它线程调用等待线程的Thread.interrupt方法来中断等待线程的等待而直接返回，这时不用获取锁，而会抛出一个InterruptedException。
- ReentrantLock.lock方法不允许Thread.interrupt中断,即使检测到Thread.isInterrupted,一样会继续尝试获取锁，失败则继续休眠。只是在最后获取锁成功后再把当前线程置为interrupted状态,然后再中断线程。
ArrayBlockingQueue的内部的信号量
ArrayBlockingQueue由于数组的容量是固定的，所以需要信号量协调put和take动作。
- 在put的时候遇到数组满的时候通过notFull.await()实现等待，直到dequeue()方法消费一个元素后执行notFull.signal()通知可以put新元素。
- 在take的时候遇到数组空的时候通过notEmpty.await()实现等待，直到enqueue()方法新增一个元素后执行notEmpty.signal()通知可以take新元素。
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
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();
}
----------------------------------------------------
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();
}
}
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}
ArrayBlockingQueue迭代器
ArrayBlockingQueue迭代器的迭代器比较简单，hasNext()判断下一个元素是否为null，next()通过移动下标获取下一个元素，中间涉及到一些下标到末尾重新从头开始。当然有一些细节代码我直接省略了不影响理解迭代过程。
public Iterator<E> iterator() {
return new Itr();
}
private class Itr implements Iterator<E> {
private int cursor;
private E nextItem;
private int nextIndex;
private E lastItem;
private int lastRet;
private int prevTakeIndex;
private int prevCycles;
private static final int NONE = -1;
private static final int REMOVED = -2;
private static final int DETACHED = -3;
Itr() {
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;
}
} finally {
lock.unlock();
}
}
private int incCursor(int index) {
// assert lock.getHoldCount() == 1;
if (++index == items.length)
index = 0;
if (index == putIndex)
index = NONE;
return index;
}
public boolean hasNext() {
// assert lock.getHoldCount() == 0;
if (nextItem != null)
return true;
noNext();
return false;
}
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;
}
}
网友评论