阻塞队列 BlockingQueue 用法详解
一、摘要
在之前的文章中,我们介绍了生产者和消费者模型的最基本实现思路,相信大家对它已经有一个初步的认识。
在 Java 的并发包里面还有一个非常重要的接口:BlockingQueue。
BlockingQueue
是一个阻塞队列,更为准确的解释是:BlockingQueue
是一个基于阻塞机制实现的线程安全的队列。通过它也可以实现生产者和消费者模型,并且效率更高、安全可靠,相比之前介绍的生产者和消费者模型,它可以同时实现生产者和消费者并行运行。
那什么是阻塞队列呢?
简单的说,就是当参数在入队和出队时,通过加锁的方式来避免线程并发操作时导致的数据异常问题。
在 Java 中,能对线程并发执行进行加锁的方式主要有synchronized
和ReentrantLock
,其中BlockingQueue
采用的是ReentrantLock
方式实现。
与此对应的还有非阻塞机制的队列,主要是采用 CAS 方式来控制并发操作,例如:ConcurrentLinkedQueue
,这个我们在后面的文章再进行分享介绍。
今天我们主要介绍BlockingQueue
相关的知识和用法,废话不多说了,进入正题!
二、BlockingQueue 方法介绍
打开BlockingQueue
的源码,你会发现它继承自Queue
,正如上文提到的,它本质是一个队列接口。
public interface BlockingQueue<E> extends Queue<E> {
//...省略
}
关于队列,我们在之前的集合系列文章中对此有过深入的介绍,本篇就再次简单的介绍一下。
队列其实是一个数据结构,元素遵循先进先出的原则,所有新元素的插入,也被称为入队操作,会插入到队列的尾部;元素的移除,也被称为出队操作,会从队列的头部开始移除,从而保证先进先出的原则。
在Queue
接口中,总共有 6 个方法,可以分为 3 类,分别是:插入、移除、查询,内容如下:
方法 | 描述 |
---|---|
add(e) | 插入元素,如果插入失败,就抛异常 |
offer(e) | 插入元素,如果插入成功,就返回 true;反之 false |
remove() | 移除元素,如果移除失败,就抛异常 |
poll() | 移除元素,如果移除成功,返回 true;反之 false |
element() | 获取队首元素,如果获取结果为空,就抛异常 |
peek() | 获取队首元素,如果获取结果为空,返回空对象 |
因为BlockingQueue
是Queue
的子接口,了解Queue
接口里面的方法,有助于我们对BlockingQueue
的理解。
除此之外,BlockingQueue
还单独扩展了一些特有的方法,内容如下:
方法 | 描述 |
---|---|
put(e) | 插入元素,如果没有插入成功,线程会一直阻塞,直到队列中有空间再继续 |
offer(e, time, unit) | 插入元素,如果在指定的时间内没有插入成功,就返回 false;反之 true |
take() | 移除元素,如果没有移除成功,线程会一直阻塞,直到队列中新的数据被加入 |
poll(time, unit) | 移除元素,如果在指定的时间内没有移除成功,就返回 false;反之 true |
drainTo(Collection c, int maxElements) | 一次性取走队列中的数据到 c 中,可以指定取的个数。该方法可以提升获取数据效率,不需要多次分批加锁或释放锁 |
分析源码,你会发现相比普通的Queue
子类,BlockingQueue
子类主要有以下几个明显的不同点:
- 1.元素插入和移除时线程安全:主要是通过在入队和出队时进行加锁,保证了队列线程安全,加锁逻辑采用
ReentrantLock
实现 - 2.支持阻塞的入队和出队方法:当队列满时,会阻塞入队的线程,直到队列不满;当队列为空时,会阻塞出队的线程,直到队列中有元素;同时支持超时机制,防止线程一直阻塞
三、BlockingQueue 用法详解
打开源码,BlockingQueue
接口的实现类非常多,我们重点讲解一下其中的 5 个非常重要的实现类,分别如下表所示。
实现类 | 功能 |
---|---|
ArrayBlockingQueue | 基于数组的阻塞队列,使用数组存储数据,需要指定长度,所以是一个有界队列 |
LinkedBlockingQueue | 基于链表的阻塞队列,使用链表存储数据,默认是一个无界队列;也可以通过构造方法中的capacity 设置最大元素数量,所以也可以作为有界队列 |
SynchronousQueue | 一种没有缓冲的队列,生产者产生的数据直接会被消费者获取并且立刻消费 |
PriorityBlockingQueue | 基于优先级别的阻塞队列,底层基于数组实现,是一个无界队列 |
DelayQueue | 延迟队列,其中的元素只有到了其指定的延迟时间,才能够从队列中出队 |
下面我们对以上实现类的用法,进行一一介绍。
3.1、ArrayBlockingQueue
ArrayBlockingQueue
是一个基于数组的阻塞队列,初始化的时候必须指定队列大小,源码实现比较简单,采用的是ReentrantLock
和Condition
实现生产者和消费者模型,部分核心源码如下:
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/** 使用数组存储队列中的元素 */
final Object[] items;
/** 使用独占锁ReetrantLock */
final ReentrantLock lock;
/** 等待出队的条件 */
private final Condition notEmpty;
/** 等待入队的条件 */
private final Condition notFull;
/** 初始化时,需要指定队列大小 */
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 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();
}
}
/**出队操作*/
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
}
ArrayBlockingQueue
采用ReentrantLock
进行加锁,只有一个ReentrantLock
对象,这意味着生产者和消费者无法并行运行。
我们看一个简单的示例代码如下:
public class Container {
/**
* 初始化阻塞队列
*/
private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
/**
* 添加数据到阻塞队列
* @param value
*/
public void add(Integer value) {
try {
queue.put(value);
System.out.println("生产者:"+ Thread.currentThread().getName()+",add:" + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 从阻塞队列获取数据
*/
public void get() {
try {
Integer value = queue.take();
System.out.println("消费者:"+ Thread.currentThread().getName()+",value:" + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 生产者
*/
public class Producer extends Thread {
private Container container;
public Producer(Container container) {
this.container = container;
}
@Override
public void run() {
for (int i = 0; i < 6; i++) {
container.add(i);
}
}
}
/**
* 消费者
*/
public class Consumer extends Thread {
private Container container;
public Consumer(Container container) {
this.container = container;
}
@Override
public void run() {
for (int i = 0; i < 6; i++) {
container.get();
}
}
}
/**
* 测试类
*/
public class MyThreadTest {
public static void main(String[] args) {
Container container = new Container();
Producer producer = new Producer(container);
Consumer consumer = new Consumer(container);
producer.start();
consumer.start();
}
}
运行结果如下:
生产者:Thread-0,add:0
生产者:Thread-0,add:1
生产者:Thread-0,add:2
生产者:Thread-0,add:3
生产者:Thread-0,add:4
生产者:Thread-0,add:5
消费者:Thread-1,value:0
消费者:Thread-1,value:1
消费者:Thread-1,value:2
消费者:Thread-1,value:3
消费者:Thread-1,value:4
消费者:Thread-1,value:5
可以很清晰的看到,生产者线程执行完毕之后,消费者线程才开始消费。
3.2、LinkedBlockingQueue
LinkedBlockingQueue
是一个基于链表的阻塞队列,初始化的时候无须指定队列大小,默认队列长度为Integer.MAX_VALUE
,也就是 int 型最大值。
同样的,采用的是ReentrantLock
和Condition
实现生产者和消费者模型,不同的是它使用了两个lock
,这意味着生产者和消费者可以并行运行,程序执行效率进一步得到提升。
部分核心源码如下:
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/** 使用出队独占锁ReetrantLock */
private final ReentrantLock takeLock = new ReentrantLock();
/** 等待出队的条件 */
private final Condition notEmpty = takeLock.newCondition();
/** 使用入队独占锁ReetrantLock */
private final ReentrantLock putLock = new ReentrantLock();
/** 等待入队的条件 */
private final Condition notFull = putLock.newCondition();
/**入队操作*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
/**出队操作*/
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
}
把最上面的样例Container
中的阻塞队列实现类换成LinkedBlockingQueue
,调整如下:
/**
* 初始化阻塞队列
*/
private final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
再次运行结果如下:
生产者:Thread-0,add:0
消费者:Thread-1,value:0
生产者:Thread-0,add:1
消费者:Thread-1,value:1
生产者:Thread-0,add:2
消费者:Thread-1,value:2
生产者:Thread-0,add:3
生产者:Thread-0,add:4
生产者:Thread-0,add:5
消费者:Thread-1,value:3
消费者:Thread-1,value:4
消费者:Thread-1,value:5
可以很清晰的看到,生产者线程和消费者线程,交替并行执行。
3.3、SynchronousQueue
SynchronousQueue
是一个没有缓冲的队列,生产者产生的数据直接会被消费者获取并且立刻消费,相当于传统的一个请求对应一个应答模式。
相比ArrayBlockingQueue
和LinkedBlockingQueue
,SynchronousQueue
实现机制也不同,它主要采用队列和栈来实现数据的传递,中间不存储任何数据,生产的数据必须得消费者处理,线程阻塞方式采用 JDK 提供的LockSupport park/unpark
函数来完成,也支持公平和非公平两种模式。
- 当采用公平模式时:使用一个 FIFO 队列来管理多余的生产者和消费者
- 当采用非公平模式时:使用一个 LIFO 栈来管理多余的生产者和消费者,这也是
SynchronousQueue
默认的模式
部分核心源码如下:
public class SynchronousQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/**不同的策略实现*/
private transient volatile Transferer<E> transferer;
/**默认非公平模式*/
public SynchronousQueue() {
this(false);
}
/**可以选策略,也可以采用公平模式*/
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
/**入队操作*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
if (transferer.transfer(e, false, 0) == null) {
Thread.interrupted();
throw new InterruptedException();
}
}
/**出队操作*/
public E take() throws InterruptedException {
E e = transferer.transfer(null, false, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
}
同样的,把最上面的样例Container
中的阻塞队列实现类换成SynchronousQueue
,代码如下:
public class Container {
/**
* 初始化阻塞队列
*/
private final BlockingQueue<Integer> queue = new SynchronousQueue<>();
/**
* 添加数据到阻塞队列
* @param value
*/
public void add(Integer value) {
try {
queue.put(value);
Thread.sleep(100);
System.out.println("生产者:"+ Thread.currentThread().getName()+",add:" + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 从阻塞队列获取数据
*/
public void get() {
try {
Integer value = queue.take();
Thread.sleep(200);
System.out.println("消费者:"+ Thread.currentThread().getName()+",value:" + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
再次运行结果如下:
生产者:Thread-0,add:0
消费者:Thread-1,value:0
生产者:Thread-0,add:1
消费者:Thread-1,value:1
生产者:Thread-0,add:2
消费者:Thread-1,value:2
生产者:Thread-0,add:3
消费者:Thread-1,value:3
生产者:Thread-0,add:4
消费者:Thread-1,value:4
生产者:Thread-0,add:5
消费者:Thread-1,value:5
可以很清晰的看到,生产者线程和消费者线程,交替串行执行,生产者每投递一条数据,消费者处理一条数据。
3.4、PriorityBlockingQueue
PriorityBlockingQueue
是一个基于优先级别的阻塞队列,底层基于数组实现,可以认为是一个无界队列。
PriorityBlockingQueue
与ArrayBlockingQueue
的实现逻辑,基本相似,也是采用ReentrantLock
来实现加锁的操作。
最大不同点在于:
- 1.
PriorityBlockingQueue
内部基于数组实现的最小二叉堆算法,可以对队列中的元素进行排序,插入队列的元素需要实现Comparator
或者Comparable
接口,以便对元素进行排序 - 2.其次,队列的长度是可扩展的,不需要显式指定长度,上限为
Integer.MAX_VALUE - 8
部分核心源码如下:
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/**队列元素*/
private transient Object[] queue;
/**比较器*/
private transient Comparator<? super E> comparator;
/**采用ReentrantLock进行加锁*/
private final ReentrantLock lock;
/**条件等待与通知*/
private final Condition notEmpty;
/**入队操作*/
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
/**出队操作*/
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
while ( (result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();
}
return result;
}
}
同样的,把最上面的样例Container
中的阻塞队列实现类换成PriorityBlockingQueue
,调整如下:
/**
* 初始化阻塞队列
*/
private final BlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
生产者插入数据的内容,我们改下插入顺序。
/**
* 生产者
*/
public class Producer extends Thread {
private Container container;
public Producer(Container container) {
this.container = container;
}
@Override
public void run() {
container.add(5);
container.add(3);
container.add(1);
container.add(2);
container.add(0);
container.add(4);
}
}
最后运行结果如下:
生产者:Thread-0,add:5
生产者:Thread-0,add:3
生产者:Thread-0,add:1
生产者:Thread-0,add:2
生产者:Thread-0,add:0
生产者:Thread-0,add:4
消费者:Thread-1,value:0
消费者:Thread-1,value:1
消费者:Thread-1,value:2
消费者:Thread-1,value:3
消费者:Thread-1,value:4
消费者:Thread-1,value:5
从日志上可以很明显看出,对于整数,默认情况下,按照升序排序,消费者默认从 0 开始处理。
3.5、DelayQueue
DelayQueue
是一个线程安全的延迟队列,存入队列的元素不会立刻被消费,只有到了其指定的延迟时间,才能够从队列中出队。
底层采用的是PriorityQueue
来存储元素,DelayQueue
的特点在于:插入队列中的数据可以按照自定义的delay
时间进行排序,快到期的元素会排列在前面,只有delay
时间小于 0 的元素才能够被取出。
部分核心源码如下:
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
/**采用ReentrantLock进行加锁*/
private final transient ReentrantLock lock = new ReentrantLock();
/**采用PriorityQueue进行存储数据*/
private final PriorityQueue<E> q = new PriorityQueue<E>();
/**条件等待与通知*/
private final Condition available = lock.newCondition();
/**入队操作*/
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
/**出队操作*/
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E first = q.peek();
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
return q.poll();
} finally {
lock.unlock();
}
}
}
同样的,把最上面的样例Container
中的阻塞队列实现类换成DelayQueue
,代码如下:
public class Container {
/**
* 初始化阻塞队列
*/
private final BlockingQueue<DelayedUser> queue = new DelayQueue<DelayedUser>();
/**
* 添加数据到阻塞队列
* @param value
*/
public void add(DelayedUser value) {
try {
queue.put(value);
System.out.println("生产者:"+ Thread.currentThread().getName()+",add:" + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 从阻塞队列获取数据
*/
public void get() {
try {
DelayedUser value = queue.take();
String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
System.out.println(time + " 消费者:"+ Thread.currentThread().getName()+",value:" + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
DelayQueue
队列中的元素需要显式实现Delayed
接口,定义一个DelayedUser
类,代码如下:
public class DelayedUser implements Delayed {
/**
* 当前时间戳
*/
private long start;
/**
* 延迟时间(单位:毫秒)
*/
private long delayedTime;
/**
* 名称
*/
private String name;
public DelayedUser(long delayedTime, String name) {
this.start = System.currentTimeMillis();
this.delayedTime = delayedTime;
this.name = name;
}
@Override
public long getDelay(TimeUnit unit) {
// 获取当前延迟的时间
long diffTime = (start + delayedTime) - System.currentTimeMillis();
return unit.convert(diffTime,TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed o) {
// 判断当前对象的延迟时间是否大于目标对象的延迟时间
return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
}
@Override
public String toString() {
return "DelayedUser{" +
"delayedTime=" + delayedTime +
", name='" + name + '\'' +
'}';
}
}
生产者插入数据的内容,做如下调整。
/**
* 生产者
*/
public class Producer extends Thread {
private Container container;
public Producer(Container container) {
this.container = container;
}
@Override
public void run() {
for (int i = 0; i < 6; i++) {
container.add(new DelayedUser(1000 * i, "张三" + i));
}
}
}
最后运行结果如下:
生产者:Thread-0,add:DelayedUser{delayedTime=0, name='张三0'}
生产者:Thread-0,add:DelayedUser{delayedTime=1000, name='张三1'}
生产者:Thread-0,add:DelayedUser{delayedTime=2000, name='张三2'}
生产者:Thread-0,add:DelayedUser{delayedTime=3000, name='张三3'}
生产者:Thread-0,add:DelayedUser{delayedTime=4000, name='张三4'}
生产者:Thread-0,add:DelayedUser{delayedTime=5000, name='张三5'}
2023-11-03 14:55:33 消费者:Thread-1,value:DelayedUser{delayedTime=0, name='张三0'}
2023-11-03 14:55:34 消费者:Thread-1,value:DelayedUser{delayedTime=1000, name='张三1'}
2023-11-03 14:55:35 消费者:Thread-1,value:DelayedUser{delayedTime=2000, name='张三2'}
2023-11-03 14:55:36 消费者:Thread-1,value:DelayedUser{delayedTime=3000, name='张三3'}
2023-11-03 14:55:37 消费者:Thread-1,value:DelayedUser{delayedTime=4000, name='张三4'}
2023-11-03 14:55:38 消费者:Thread-1,value:DelayedUser{delayedTime=5000, name='张三5'}
可以很清晰的看到,延迟时间最低的排在最前面。
四、小结
最后我们来总结一下BlockingQueue
阻塞队列接口,它提供了很多非常丰富的生产者和消费者模型的编程实现,同时兼顾了线程安全和执行效率的特点。
开发者可以通过BlockingQueue
阻塞队列接口,简单的代码编程即可实现多线程中数据高效安全传输的目的,确切的说,它帮助开发者减轻了不少的编程难度。
在实际的业务开发中,其中LinkedBlockingQueue
使用的是最广泛的,因为它的执行效率最高,在使用的时候,需要平衡好队列长度,防止过大导致内存溢出。
举个最简单的例子,比如某个功能上线之后,需要做下压力测试,总共需要请求 10000 次,采用 100 个线程去执行,测试服务是否能正常工作。如何实现呢?
可能有的同学想到,每个线程执行 100 次请求,启动 100 个线程去执行,可以是可以,就是有点笨拙。
其实还有另一个办法,就是将 10000 个请求对象,存入到阻塞队列中,然后采用 100 个线程去消费执行,这种编程模型会更佳灵活。
具体示例代码如下:
public static void main(String[] args) throws InterruptedException {
// 将每个用户访问百度服务的请求任务,存入阻塞队列中
// 也可以也采用多线程写入
BlockingQueue<String> queue = new LinkedBlockingQueue<>();
for (int i = 0; i < 10000; i++) {
queue.put("https://www.baidu.com?paramKey=" + i);
}
// 模拟100个线程,执行10000次请求访问百度
final int threadNum = 100;
for (int i = 0; i < threadNum; i++) {
final int threadCount = i + 1;
new Thread(new Runnable() {
@Override
public void run() {
System.out.println("thread " + threadCount + " start");
boolean over = false;
while (!over) {
String url = queue.poll();
if(Objects.nonNull(url)) {
// 发起请求
String result =HttpUtils.getUrl(url);
System.out.println("thread " + threadCount + " run result:" + result);
}else {
// 任务结束
over = true;
System.out.println("thread " + threadCount + " final");
}
}
}
}).start();
}
}
本文主要围绕BlockingQueue
阻塞队列接口,从方法介绍到用法详解,做了一次知识总结,如果有描述不对的地方,欢迎留言指出!
五、参考
1、https://www.cnblogs.com/xrq730/p/4855857.html
2、https://juejin.cn/post/6999798721269465102
作者:潘志的技术笔记
出处:https://pzblog.cn/
版权归作者所有,转载请注明出处