简介
BlockingQueue 是 Java concurrent包提供的多线程安全的阻塞队列,其子类包括 LinkedBlockingQueue 和 ArrayBlockingQueue。
关键API
说到队列,自然少不了首尾的插入删除操作,BlockingQueue的API中提供了好几种插入删除方法。
这些方法在遇到无法满足的执行条件时,如队列满了(添加元素时)/队列为空(取出元素时),会采取不同的措施:抛出异常,返回false/null,阻塞调用API的线程,等待一定时间等。具体如下表:
Throws exception | Special value | Blocks | Times out | |
Insert | add(e) | offer(e) | put(e) | offer(e,time,unit) |
Remove | remove() | poll() | take() | poll(time,unit) |
Examine | element() | peek() | not applicable | not applicable |
ArrayBlockingQueue
ArrayBlockingQueue是一个基于数组的阻塞队列,在创建一个ArrayBlockingQueue时,需要提供的一个表示队列大小的参数。
ArrayBlockingQueue是线程安全的,但这是怎么做到的呢?这需要注意类中的三个属性:
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
这里一个锁,两个条件变量,管理所有使用API的线程的互斥和同步。具体的锁和条件变量的理论知识,可以参见相关的操作系统书籍。
offer vs put
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
insert(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();
insert(e);
} finally {
lock.unlock();
}
}
可以看到,两个方法中在往队列里添加元素时,都是先对临界区加锁,不同在于,offer方法中若是检测出队列已满,会直接返回false;put方法中,若是检测出队列已满,线程会在 notFull 条件变量中阻塞,这样线程会释放锁,让其他线程进入临界区。以后某个时间,一个线程从队列中取出元素,队列不再为空,并且该线程唤醒在notFull条件变量上阻塞的线程,put方法才有可能完成。
poll vs take
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return extract();
} finally {
lock.unlock();
}
}
take的逻辑和put差不多,只不过这次是若队列为空,则线程阻塞在notEmpty 条件变量上,等待其他线程往队列中添加元素,并唤醒在notEmpty上阻塞的线程。
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : extract();
} finally {
lock.unlock();
}
}
poll方法在检测到队列为空时,直接返回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 extract();
} finally {
lock.unlock();
}
}
poll 还有个比较有趣的重载实现,这里利用Condition变量的计时器方法awaitNanos 。先将时间大小根据时间单位换算成纳秒的数值,当队列容量为0是,使用Condition.awaitNanos(…),进行计时,超时后返回null。