顾名思义,ArrayBlockingQueue是基于数组实现的有界阻塞队列。该队列对元素进行FIFO排序。队列的首元素是在该队列中驻留时间最长的元素。队列的尾部是在该队列中停留时间最短的元素。新的元素被插入到队列的尾部,队列检索操作获取队列头部的元素。
ArrayBlockingQueue是一个经典的“有界缓冲区(bounded buffer)”,其中内部包含了一个固定大小的数组,用于承载包含生产者插入的和消费者提取的元素。ArrayBlockingQueue的容量一旦创建,不可更改。试图将一个元素放入一个满队列将导致操作阻塞;试图从空队列中取出一个元素也同样会阻塞。
ArrayBlockingQueue支持排序的可选公平策略,用于等待生产者和消费者线程。默认情况下,不保证此顺序。然而,一个由公平性设置为true构造的队列允许线程以FIFO顺序访问。公平性一般会降低吞吐量,但可以减少可变性,避免线程饿死。
ArrayBlockingQueue类及其迭代器实现了Collection和Iterator接口的所有可选方法。ArrayBlockingQueue是Java Collections Framework的一个成员。
1. ArrayBlockingQueue的声明
ArrayBlockingQueue的接口和继承关系如下
publicclassArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
…
}
完整的接口继承关系如下图所示。
从上述代码可以看出,ArrayBlockingQueue既实现了BlockingQueue<E>和java.io.Serializable接口,又继承了java.util.AbstractQueue<E>。其中,AbstractQueue是Queue接口的抽象类,核心代码如下。
package java.util;
publicabstractclass AbstractQueue<E>
extends AbstractCollection<E>
implements Queue<E> {
protected AbstractQueue() {
}
publicboolean add(E e) {
if (offer(e))
returntrue;
else
thrownew IllegalStateException("Queue full");
}
public E remove() {
E x = poll();
if (x != null)
returnx;
else
thrownew NoSuchElementException();
}
public E element() {
E x = peek();
if (x != null)
returnx;
else
thrownew NoSuchElementException();
}
publicvoid clear() {
while (poll() != null)
;
}
publicboolean addAll(Collection<? extends E> c) {
if (c == null)
thrownew NullPointerException();
if (c == this)
thrownew IllegalArgumentException();
booleanmodified = false;
for (E e : c)
if (add(e))
modified = true;
returnmodified;
}
}
2. ArrayBlockingQueue的成员变量和构造函数
以下是ArrayBlockingQueue的构造函数和成员变量。
// 元素数组
final Object[] items;
// 消费索引,用于take、poll、peek或remove操作
inttakeIndex;
// 生产索引,用于put、offer或add操作
intputIndex;
// 队列中的元素个数
intcount;
/*
* 使用经典的双条件算法(two-condition algorithm)实现并发控制
*/
// 操作数组确保原子性的锁
finalReentrantLocklock;
// 数组非空,唤醒消费者
privatefinal Condition notEmpty;
// 数组非满,唤醒生产者
privatefinal Condition notFull;
// 迭代器状态
transient Itrs itrs;
public ArrayBlockingQueue(intcapacity) {
this(capacity, false);
}
public ArrayBlockingQueue(intcapacity, booleanfair) {
if (capacity <= 0)
thrownew IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
public ArrayBlockingQueue(intcapacity, booleanfair,
Collection<? extends E> c) {
this(capacity, fair);
final ReentrantLock lock = this.lock;
lock.lock(); // 只锁可见,不互斥
try {
final Object[] items = this.items;
inti = 0;
try {
for (E e : c)
items[i++] = Objects.requireNonNull(e);
} catch (ArrayIndexOutOfBoundsException ex) {
thrownew IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock(); // 解锁
}
}
从上述代码可以看出,构造函数有三种。构造函数中的参数含义如下
l capacity用于设置队列容量
l fair用于设置访问策略,如果为true,则对线程的访问在插入或移除时被阻塞,则按FIFO顺序处理;如果为false,则访问顺序未指定
l c用于设置最初包含给定集合的元素,按集合迭代器的遍历顺序添加
类成员items是一个数组,用于存储队列中的元素。关键字final指明了,当ArrayBlockingQueue构造完成之后,通过new Object[capacity]的方式初始化items数组完成后,则后续items的容量将不再变化。
访问策略是通过ReentrantLock来实现的。通过两个加锁条件notEmpty、notFull来实现并发控制。这是典型的双条件算法(two-condition algorithm)。
ArrayBlockingQueue生产则增加putIndex,消费则增加takeIndex。
Itrs用于记录当前活动迭代器的共享状态,如果已知不存在任何迭代器,则为null。允许队列操作更新迭代器状态。迭代器状态不是本节的重点,不再深入探讨。
3. ArrayBlockingQueue的核心方法
以下对ArrayBlockingQueue常用核心方法的实现原理进行解释。
3.1. offer(e)
执行offer(e)方法后有两种结果
l 队列未满时,返回 true
l 队列满时,返回 false
ArrayBlockingQueue的offer (e)方法源码如下:
publicbooleanoffer(E e) {
Objects.requireNonNull(e);
final ReentrantLock lock = this.lock;
lock.lock(); // 加锁
try {
if (count == items.length)
returnfalse;
else {
enqueue(e); // 入队
returntrue;
}
} finally {
lock.unlock(); // 解锁
}
}
从上面代码可以看出,执行offer(e)方法时,分为以下几个步骤:
l 为了确保并发操作的安全先做了加锁处理。
l 而后判断count是否与数组items的长度一致,如果一致则证明队列已经满了,直接返回false;否则执行enqueue(e)方法做元素的入队,并返回true。
l 最后解锁。
enqueue(e)方法源码如下:
privatevoidenqueue(E e) {
final Object[] items = this.items;
items[putIndex] = e;
if (++putIndex == items.length) putIndex = 0;
count++;
notEmpty.signal(); // 唤醒等待中的线程
}
上面代码比较简单,在当前索引(putIndex)位置放置待入队的元素,而后putIndex和count分别递增,并通过signal()方法唤醒等待中的线程。其中一个注意点是,当putIndex 等于数组items长度时,putIndex置为0。
思考:当putIndex 等于数组items长度时,putIndex为什么置为0呢?
3.2. put(e)
执行put(e)方法后有两种结果:
•
l 队列未满时,直接插入没有返回值
l 队列满时,会阻塞等待,一直等到队列未满时再插入
ArrayBlockingQueue的put (e)方法源码如下:
publicvoidput(E e) throws InterruptedException {
Objects.requireNonNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 获取锁
try {
while (count == items.length)
notFull.await(); // 使线程等待
enqueue(e); // 入队
} finally {
lock.unlock(); // 解锁
}
}
从上面代码可以看出,put(e)方法的实现,分为以下几个步骤:
l 先是要获取锁。
l 而后判断count是否与数组items的长度一致,如果一致则证明队列已经满了,就等待;否则执行enqueue(e)方法做元素的入队。
l 最后解锁。
3.3. offer(e,time,unit)
offer(e,time,unit)方法与offer(e)方法不同之处在于,前者加入了等待机制。设定等待的时间,如果在指定时间内还不能往队列中插入数据则返回false。执行offer(e,time,unit)方法有两种结果:
•
l 队列未满时,返回 true
l 队列满时,会阻塞等待,如果在指定时间内还不能往队列中插入数据则返回 false
ArrayBlockingQueue的put (e)方法源码如下:
publicboolean offer(E e, longtimeout, TimeUnit unit)
throws InterruptedException {
Objects.requireNonNull(e);
longnanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 获取锁
try {
while (count == items.length) {
if (nanos <= 0L)
returnfalse;
nanos = notFull.awaitNanos(nanos); // 使线程等待指定的时间
}
enqueue(e);
returntrue;
} finally {
lock.unlock(); // 解锁
}
}
从上面代码可以看出,offer(e,time,unit)方法的实现,分为以下几个步骤:
l 先是要获取锁。
l 而后判断count是否与数组items的长度一致,如果一致则证明队列已经满了,就等待;否则执行enqueue(e)方法做元素的入队。
l 最后解锁。
3.4. add(e)
执行add(e)方法后有两种结果
l 队列未满时,返回 true
l 队列满时,则抛出异常
ArrayBlockingQueue的add(e)方法源码如下:
publicbooleanadd(E e) {
returnsuper.add(e);
}
从上面代码可以看出,add(e)方法的实现,直接是调用了父类AbstractQueue的add(e)方法。而AbstractQueue的add(e)方法源码如下:
publicbooleanadd(E e) {
if (offer(e))
returntrue;
else
thrownew IllegalStateException("Queue full");
}
从上面代码可以看出,add(e)方法又调用了offer(e)方法。offer(e)方法此处不再赘述。
3.5. poll ()
执行poll ()方法后有两种结果:
l 队列不为空时,返回队首值并移除
l 队列为空时,返回 null
ArrayBlockingQueue的poll()方法源码如下:
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock(); // 加锁
try {
return (count == 0) ? null : dequeue(); // 出队
} finally {
lock.unlock(); // 解锁
}
}
从上面代码可以看出,执行poll()方法时,分为以下几个步骤:
l 为了确保并发操作的安全先做了加锁处理。
l 而后判断count是否等于0,如果等于0则证明队列为空,直接返回null;否则执行dequeue()方法做元素的出队。
l 最后解锁。
dequeue()方法源码如下:
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E e = (E) items[takeIndex];
items[takeIndex] = null; // 删除数据
if (++takeIndex == items.length) takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal(); // 唤醒等待中的线程
returne;
}
上面代码比较简单,在当前索引(takeIndex)位置取出待出队的元素并删除队列中的元素,而后takeIndex递增count递减,并通过signal()方法唤醒等待中的线程。其中一个注意点是,当takeIndex等于数组items长度时,takeIndex置为0。
3.6. take()
执行take()方法后有两种结果:
l 队列不为空时,返回队首值并移除
l 队列为空时,会阻塞等待,一直等到队列不为空时再返回队首值
ArrayBlockingQueue的take ()方法源码如下:
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 获取锁
try {
while (count == 0)
notEmpty.await(); // 使线程等待
return dequeue(); // 出队
} finally {
lock.unlock(); // 解锁
}
}
从上面代码可以看出,执行take()方法时,分为以下几个步骤:
l 先是要获取锁。
l 而后判断count是否等于0,如果等于0则证明队列为空,会阻塞等待;否则执行dequeue()方法做元素的出队。
l 最后解锁。
dequeue()方法此处不再赘述。
3.7. poll(time,unit)
poll(time,unit)方法与poll()方法不同之处在于,前者加入了等待机制。设定等待的时间,如果在指定时间内队列还为空,则返回null。执行poll(time,unit)方法后有两种结果:
l 队列不为空时,返回队首值并移除
l 队列为空时,会阻塞等待,如果在指定时间内队列还为空则返回 null
ArrayBlockingQueue的poll(time,unit)方法源码如下:
public E poll(longtimeout, TimeUnit unit) throws InterruptedException {
longnanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 获取锁
try {
while (count == 0) {
if (nanos <= 0L)
returnnull;
nanos = notEmpty.awaitNanos(nanos); // 使线程等待指定的时间
}
return dequeue(); // 出队
} finally {
lock.unlock(); // 解锁
}
}
从上面代码可以看出,执行poll(time,unit)方法时,分为以下几个步骤:
l 先是要获取锁。
l 而后判断count是否等于0,如果等于0则证明队列为空,会阻塞等待;否则执行dequeue()方法做元素的出队。
l 最后解锁。
dequeue()方法此处不再赘述。
3.8. remove()
执行remove()方法后有两种结果:
l 队列不为空时,返回队首值并移除
l 队列为空时,抛出异常
ArrayBlockingQueue的remove()方法其实是调用了父类AbstractQueue的remove()方法,源码如下:
public E remove() {
E x = poll();
if (x != null)
returnx;
else
thrownew NoSuchElementException();
}
从上面代码可以看出,remove()直接调用了poll()方法。如果poll()方法返回结果为null,则抛出NoSuchElementException异常。
poll()方法此处不再赘述。
3.9. peek()
执行peek()方法后有两种结果:
l 队列不为空时,返回队首值但不移除
l 队列为空时,返回null
peek()方法源码如下:
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock(); // 加锁
try {
return itemAt(takeIndex); // 空则返回null
} finally {
lock.unlock(); // 解锁
}
}
final E itemAt(inti) {
return (E) items[i];
}
从上面代码可以看出,peek()方法比较简单,直接就是获取了数组里面的索引为takeIndex的元素。
3.10. element()
执行element()方法后有两种结果:
l 队列不为空时,返回队首值但不移除
l 队列为空时,抛出异常
element()方法其实是调用了父类AbstractQueue的element()方法,源码如下:
public E element() {
E x = peek();
if (x != null)
returnx;
else
thrownew NoSuchElementException();
}
从上面代码可以看出,执行element()方法时,先是获取peek()方法的结果,如果结果是null,则抛出NoSuchElementException异常。
4. ArrayBlockingQueue的单元测试
ArrayBlockingQueue的单元测试如下:
package com.waylau.java.demo.datastructure;
importstatic org.junit.jupiter.api.Assertions.assertEquals;
importstatic org.junit.jupiter.api.Assertions.assertFalse;
importstatic org.junit.jupiter.api.Assertions.assertNotNull;
importstatic org.junit.jupiter.api.Assertions.assertNull;
importstatic org.junit.jupiter.api.Assertions.assertThrows;
importstatic org.junit.jupiter.api.Assertions.assertTrue;
import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
/**
* ArrayBlockingQueue Tests
*
* @since 1.0.0 2020年5月3日
* @author <a href="https://waylau.com">Way Lau</a>
*/
class ArrayBlockingQueueTests {
@Test
void testOffer() {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列未满时,返回 true
booleanresultNotFull = queue.offer("Java");
assertTrue(resultNotFull);
// 测试队列满则,返回 false
queue.offer("C");
queue.offer("Python");
booleanresultFull = queue.offer("C++");
assertFalse(resultFull);
}
@Test
void testPut() throws InterruptedException {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列未满时,直接插入没有返回值;
queue.put("Java");
// 测试队列满则,会阻塞等待,一直等到队列未满时再插入。
queue.put("C");
queue.put("Python");
queue.put("C++"); // 阻塞等待
}
@Test
void testOfferTime() throws InterruptedException {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列未满时,返回 true
booleanresultNotFull = queue.offer("Java", 5, TimeUnit.SECONDS);
assertTrue(resultNotFull);
// 测试队列满则,返回 false
queue.offer("C");
queue.offer("Python");
booleanresultFull = queue.offer("C++", 5, TimeUnit.SECONDS); // 等5秒
assertFalse(resultFull);
}
@Test
void testAdd() {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列未满时,返回 true
booleanresultNotFull = queue.add("Java");
assertTrue(resultNotFull);
// 测试队列满则抛出异常
queue.add("C");
queue.add("Python");
Throwable excpetion = assertThrows(IllegalStateException.class, () -> {
queue.add("C++");// 抛异常
});
assertEquals("Queue full", excpetion.getMessage());
}
@Test
void testPoll() throws InterruptedException {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列为空时,返回 null
String resultEmpty = queue.poll();
assertNull(resultEmpty);
// 测试队列不为空时,返回队首值并移除
queue.put("Java");
queue.put("C");
queue.put("Python");
String resultNotEmpty = queue.poll();
assertEquals("Java", resultNotEmpty);
}
@Test
void testTake() throws InterruptedException {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列不为空时,返回队首值并移除
queue.put("Java");
queue.put("C");
queue.put("Python");
String resultNotEmpty = queue.take();
assertEquals("Java", resultNotEmpty);
// 测试队列为空时,会阻塞等待,一直等到队列不为空时再返回队首值
queue.clear();
String resultEmpty = queue.take(); // 阻塞等待
assertNotNull(resultEmpty);
}
@Test
void testPollTime() throws InterruptedException {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列不为空时,返回队首值并移除
queue.put("Java");
queue.put("C");
queue.put("Python");
String resultNotEmpty = queue.poll(5, TimeUnit.SECONDS);
assertEquals("Java", resultNotEmpty);
// 测试队列为空时,会阻塞等待,如果在指定时间内队列还为空则返回 null
queue.clear();
String resultEmpty = queue.poll(5, TimeUnit.SECONDS); // 等待5秒
assertNull(resultEmpty);
}
@Test
void testRemove() throws InterruptedException {
// 初始化队列
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列为空时,抛出异常
Throwable excpetion = assertThrows(NoSuchElementException.class, () -> {
queue.remove();// 抛异常
});
assertEquals(null, excpetion.getMessage());
// 测试队列不为空时,返回队首值并移除
queue.put("Java");
queue.put("C");
queue.put("Python");
String resultNotEmpty = queue.remove();
assertEquals("Java", resultNotEmpty);
}
@Test
void testPeek() throws InterruptedException {
// 初始化队列
Queue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列不为空时,返回队首值并但不移除
queue.add("Java");
queue.add("C");
queue.add("Python");
String resultNotEmpty = queue.peek();
assertEquals("Java", resultNotEmpty);
resultNotEmpty = queue.peek();
assertEquals("Java", resultNotEmpty);
resultNotEmpty = queue.peek();
assertEquals("Java", resultNotEmpty);
// 测试队列为空时,返回null
queue.clear();
String resultEmpty = queue.peek();
assertNull(resultEmpty);
}
@Test
void testElement() throws InterruptedException {
// 初始化队列
Queue<String> queue = new ArrayBlockingQueue<String>(3);
// 测试队列不为空时,返回队首值并但不移除
queue.add("Java");
queue.add("C");
queue.add("Python");
String resultNotEmpty = queue.element();
assertEquals("Java", resultNotEmpty);
resultNotEmpty = queue.element();
assertEquals("Java", resultNotEmpty);
resultNotEmpty = queue.element();
assertEquals("Java", resultNotEmpty);
// 测试队列为空时,抛出异常
queue.clear();
Throwable excpetion = assertThrows(NoSuchElementException.class, () -> {
queue.element();// 抛异常
});
assertEquals(null, excpetion.getMessage());
}
}
5. ArrayBlockingQueue的应用案例
以下是一个生产者-消费者的示例。该示例模拟了1个生产者,2个消费者。当队列满时,则会阻塞生产者生产;当队列空时,则会阻塞消费者消费。
package com.waylau.java.demo.datastructure;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
* ArrayBlockingQueue Demo
*
* @since 1.0.0 2020年5月3日
* @author <a href="https://waylau.com">Way Lau</a>
*/
publicclass ArrayBlockingQueueDemo {
publicstaticvoid main(String[] args) {
BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
// 1个生产者
Producer p = new Producer(queue);
// 2个消费者
Consumer c1 = new Consumer("c1", queue);
Consumer c2 = new Consumer("c2", queue);
// 启动线程
new Thread(p).start();
new Thread(c1).start();
new Thread(c2).start();
}
}
class Producer implements Runnable {
privatefinal BlockingQueue<String> queue;
Producer(BlockingQueue<String> queue) {
this.queue = queue;
}
publicvoid run() {
try {
while (true) {
// 模拟耗时操作
Thread.sleep(1000L);
queue.put(produce());
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
String produce() {
String apple = "apple: " + System.currentTimeMillis();
System.out.println("produce " + apple);
returnapple;
}
}
class Consumer implements Runnable {
privatefinal BlockingQueue<String> queue;
privatefinal String name;
Consumer(String name, BlockingQueue<String> queue) {
this.queue = queue;
this.name = name;
}
publicvoid run() {
try {
while (true) {
// 模拟耗时操作
Thread.sleep(2000L);
consume(queue.take());
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
void consume(Object x) {
System.out.println(this.name + " consume " + x);
}
}
运行上述程序,输出内容如下:
produce apple: 1590308383034
c2 consume apple: 1590308383034
produce apple: 1590308384034
c1 consume apple: 1590308384034
produce apple: 1590308385036
c2 consume apple: 1590308385036
produce apple: 1590308386036
c1 consume apple: 1590308386036
produce apple: 1590308387036
c2 consume apple: 1590308387036
produce apple: 1590308388036
c1 consume apple: 1590308388036
produce apple: 1590308389041
c2 consume apple: 1590308389041
produce apple: 1590308390041
c1 consume apple: 1590308390041
produce apple: 1590308391042
c2 consume apple: 1590308391042
produce apple: 1590308392042
c1 consume apple: 1590308392042
6. 参考引用
本系列归档至《Java数据结构及算法实战》:https://github.com/waylau/java-data-structures-and-algorithms-in-action
《数据结构和算法基础(Java语言实现)》(柳伟卫著,北京大学出版社出版):https://item.jd.com/13014179.html