在Java中如何使用BlockingQueue实现生产者消费者模式
BlockingQueue是实现生产者消费者模式的线程安全方式，其内部处理同步问题，生产者调用put阻塞入队，消费者调用take阻塞出队，队列满时生产者等待，队列空时消费者等待。
在Java中，BlockingQueue 是实现生产者消费者模式最简单且线程安全的方式。它内部已经处理了线程同步问题，生产者线程向队列添加元素，消费者线程从队列取出元素，当队列满时生产者自动阻塞，队列空时消费者自动阻塞。
1. BlockingQueue 的基本特性
BlockingQueue 是 java.util.concurrent 包中的接口，常用实现类有：
- ArrayBlockingQueue：有界阻塞队列，基于数组实现
- LinkedBlockingQueue：可选有界队列，基于链表实现
- PriorityBlockingQueue：支持优先级的无界阻塞队列
- SynchronousQueue：不存储元素的阻塞队列，每个插入必须等待对应移除
关键方法：
- put(E e)：将元素插入队列，如果队列满则阻塞
- take()：从队列获取并移除元素，如果队列空则阻塞
- offer(E e, long timeout, TimeUnit unit)：带超时的插入
- poll(long timeout, TimeUnit unit)：带超时的取出
2. 实现生产者消费者模型
下面是一个使用 ArrayBlockingQueue 的简单示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
// 共享数据
class Data {
private final int value;
public Data(int value) {
this.value = value;
}
@Override
public String toString() {
return "Data{" + "value=" + value + '}';
}
}
// 生产者
class Producer implements Runnable {
private final BlockingQueue queue;
public Producer(BlockingQueue
queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 1; i <= 5; i++) {
Data data = new Data(i);
queue.put(data); // 阻塞插入
System.out.println("生产者生产: " + data);
Thread.sleep(500); // 模拟耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
// 消费者
class Consumer implements Runnable {
private final BlockingQueue queue;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (true) {
Data data = queue.take(); // 阻塞取出
System.out.println("消费者消费: " + data);
Thread.sleep(1000); // 消费较慢
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
// 主程序
public class ProducerConsumerExample {
public static void main(String[] args) {
// 创建容量为3的阻塞队列
BlockingQueue queue = new ArrayBlockingQueue<>(3);
// 启动生产者和消费者线程
Thread producerThread = new Thread(new Producer(queue));
Thread consumerThread = new Thread(new Consumer(queue));
producerThread.start();
consumerThread.start();
// 可以设置运行时间后中断
try {
Thread.sleep(6000);
} catch (InterruptedException e) {
e.printStackTrace();
}
producerThread.interrupt();
consumerThread.interrupt();
}
}
queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 1; i <= 5; i++) {
Data data = new Data(i);
queue.put(data); // 阻塞插入
System.out.println("生产者生产: " + data);
Thread.sleep(500); // 模拟耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
// 消费者
class Consumer implements Runnable {
private final BlockingQueue queue;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (true) {
Data data = queue.take(); // 阻塞取出
System.out.println("消费者消费: " + data);
Thread.sleep(1000); // 消费较慢
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
// 主程序
public class ProducerConsumerExample {
public static void main(String[] args) {
// 创建容量为3的阻塞队列
BlockingQueue queue = new ArrayBlockingQueue<>(3);
// 启动生产者和消费者线程
Thread producerThread = new Thread(new Producer(queue));
Thread consumerThread = new Thread(new Consumer(queue));
producerThread.start();
consumerThread.start();
// 可以设置运行时间后中断
try {
Thread.sleep(6000);
} catch (InterruptedException e) {
e.printStackTrace();
}
producerThread.interrupt();
consumerThread.interrupt();
}
}
3. 关键点说明
使用 BlockingQueue 实现生产者消费者模式时需要注意以下几点：
- 不需要手动使用 synchronized 或 wait/notify，BlockingQueue 内部已实现线程安全
- put 和 take 方法会自动阻塞，适合实时性要求不高的场景
- 选择合适的队列大小，避免内存浪费或频繁阻塞
- 合理处理 InterruptedException，通常通过中断线程退出循环
- 多个生产者和消费者可以共享同一个队列，无需额外同步
基本上就这些。BlockingQueue 封装了复杂的线程协作逻辑，让开发者能更专注于业务本身。只要理解 put/take 的阻塞性质，就能轻松构建稳定高效的生产者消费者系统。
