介绍阻塞队列 BlockingQueue 类
介绍阻塞队列 BlockingQueue 类
本文我们认识java.util.concurrent包中非常有用解决并发生产消费问题的类。我们将学习BlockingQueue接口的API以及如何利用这些方法使并发变成更简单。文章最后我们展示一个简单示例包括多线程环境下的生产者和消费者。
BlockingQueue 类型
我们需要区分两种类型的BlockingQueue：
- unbounded queue – 可以无限增长
- bounded queue – 指定最大容量
Unbounded Queue
创建Unbounded Queue很简单：
BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>();blockingQueue队列的容量是Integer.MAX_VALUE。所有给Unbounded Queue增加的操作都不会阻塞，因为其容量可以无限增长。
使用Unbounded Queue设计生产者和消费者程序时，最重要的要让消费者能够尽快消费生产者往队列中增加的消息，否则内存会填满并出现OutOfMemory异常。
Bounded Queue
第二种队列是Bounded Queue，通过给构造函数传递容量参数进行创建：
BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>(10);这里我们创建了容量为10的blockingQueue队列，这意味着生产者给已经满了的队列增加元素时，根据使用增加方法（offer,add,put)的不同，会阻塞直到有多余的空间才会插入或操作失败。
使用Bounded Queue是设计并发编程的很好的方式，因为往满的队列中增加新元素时，操作需要等待直到消费者消费消息使队列有多余空间，这使我们无需任何额外努力就能实现节流。
BlockingQueue API
BlockingQueue 接口有两类方法————负责插入队列元素的方法以及从队列中获取元素的方法。这两组方法行为差异在于队列满的和空的情况。
插入元素
- add() – 如果插入成功返回true，反之抛出 IllegalStateException异常
- put() – 往队列中插入特定元素，需要等待空闲空间
- offer() – 如果插入成功返回true，反之为false
- offer(E e, long timeout, TimeUnit unit) – 尝试插入元素至队列，在特定的时间内等待空闲空间
获取元素
take() – 等待获取队列首元素并删除该元素，如果队列为空，阻塞并等待队列中有元素
poll(long timeout, TimeUnit unit) – 获取并删除队列首元素,如果队列为空则等待特定时间，超时返回null
构建生产者消费者程序，这些BlockingQueue接口的方法实现阻塞很重要。
多线程生产者-消费者示例
该示例有两部分组成——生产者和消费者。
生产者产生0~100之间的随机数并插入队列。可以有多个线程使用put方法插入元素，队列没有空间会阻塞。
特别重要的是，我们需要停止消费者线程，避免无限地等待队列有元素。
比较好的技术是从生产者发送信号到消费者，当没有消息需要生产时，发送一个特定的消息poison pill（毒药），当消费者从队列总接收该消息则执行完成。
下面看下生产者类：
public class NumbersProducer implements Runnable {
private BlockingQueue<Integer> numbersQueue;
// 特殊值视为毒药（结束标志）
private final int poisonPill;
// 生产毒药数量
private final int poisonPillPerProducer;
public NumbersProducer(BlockingQueue<Integer> numbersQueue, int poisonPill, int poisonPillPerProducer) {
this.numbersQueue = numbersQueue;
this.poisonPill = poisonPill;
this.poisonPillPerProducer = poisonPillPerProducer;
}
public void run() {
try {
generateNumbers();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void generateNumbers() throws InterruptedException {
for (int i = 0; i < 100; i++) {
numbersQueue.put(ThreadLocalRandom.current().nextInt(100));
}
for (int j = 0; j < poisonPillPerProducer; j++) {
numbersQueue.put(poisonPill);
}
}
}生产者构造函数带BlockingQueue参数用于协调生产者和消费者之间的处理过程。generateNumbers() 方法给队列中插入100个元素，同时也需要毒药信息，使消费者知道什么时间结束执行。该消息需要插入到队列poisonPillPerProducer次。
每个消费者使用take方法从BlockingQueue队列取一个元素，如果队列没有元素对阻塞。取得消息后需要判断是否为毒药消息，如果是则结束线程的执行，否则在控制台打印包括当前线程名称及消息。
下面看消费者是如何工作：
public class NumbersConsumer implements Runnable {
private BlockingQueue<Integer> queue;
// 毒药标志
private final int poisonPill;
public NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) {
this.queue = queue;
this.poisonPill = poisonPill;
}
public void run() {
try {
while (true) {
Integer number = queue.take();
if (number.equals(poisonPill)) {
return;
}
System.out.println(Thread.currentThread().getName() + " result: " + number);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}特别需要注意的是队列的使用。与生产者构造函数一样，消费者也传入队列作为参数。因为BlockingQueue可以在线程间共享，无需显示使用synchronization关键字。
现在生产者和消费者都有了，我们需要定义队列测试程序，这里定义队列容量为10.
我们可以定义多个生产者线程，以及与处理器数相同数量的消费者线程。
int BOUND = 10;
int N_PRODUCERS = 4;
int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
int poisonPill = Integer.MAX_VALUE;
int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS;
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(BOUND);
for (int i = 1; i < N_PRODUCERS; i++) {
new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer)).start();
}
for (int j = 0; j < N_CONSUMERS; j++) {
new Thread(new NumbersConsumer(queue, poisonPill)).start();
}使用带容量的参数创建BlockingQueue，我们创建4个生产者和N个消费者。我们创建4个生产者和N个消费者，并指定Integer.MAX_VALUE为结束信号（毒药消息），因为生产者在正常工作条件下不会发送这样消息。最重要的是这里使用BlockingQueue协调两者。
当运行程序时，4个生产者线程产生随机数插入BlockingQueue队列，消费者从队列中获取消息。每个线程在控制台打印线程名称及消息。
总结
本文通过示例展示如何使用BlockingQueue构建多线程环境下生产者和消费者协作程序，并解释队列的插入和获取方法。
手机扫一扫
移动阅读更方便
你可能感兴趣的文章