如何在Java中使用BlockingQueue实现生产者消费者
BlockingQueue是实现生产者消费者模型的线程安全方式，其put()和take()方法自动处理阻塞。1. 特性：BlockingQueue为接口，常用实现有ArrayBlockingQueue（有界数组队列）、LinkedBlockingQueue（可选有界链表队列）和SynchronousQueue（不存储元素的同步队列）。2. 示例：通过SharedData封装BlockingQueue，生产者调用put()添加数据，消费者调用take()获取数据，队列满时生产者阻塞，队列空时消费者阻塞，实现高效线程协作。
在Java中，BlockingQueue 是实现生产者消费者模型最简单且线程安全的方式。它内部已经处理了线程同步问题，生产者线程向队列添加元素，消费者线程从队列获取元素，当队列满时生产者自动阻塞，队列空时消费者自动阻塞。
1. BlockingQueue 的基本特性
BlockingQueue 是 java.util.concurrent 包中的接口，常见实现类有：
- ArrayBlockingQueue：有界阻塞队列，基于数组实现
- LinkedBlockingQueue：可选有界队列，基于链表实现
- SynchronousQueue：不存储元素的阻塞队列，每个插入必须等待对应移除
关键方法说明：
- put(e)：将元素放入队列，如果队列满则阻塞
- take()：从队列取出一个元素，如果队列空则阻塞
- offer(e, time, unit) / poll(time, unit)：带超时的插入和取出
2. 实现生产者消费者示例
以下是一个使用 ArrayBlockingQueue 的完整示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
// 共享数据
class SharedData {
private final BlockingQueue queue;
private final int capacity;
public SharedData(int capacity) {
this.queue = new ArrayBlockingQueue<>(capacity);
this.capacity = capacity;
}
public void produce(int value) throws InterruptedException {
queue.put(value);
System.out.println("生产：" + value + "，队列大小：" + queue.size());
}
public int consume() throws InterruptedException {
int value = queue.take();
System.out.println("消费：" + value + "，队列大小：" + queue.size());
return value;
}
}
// 生产者线程
class Producer implements Runnable {
private final SharedData data;
private final int maxItems;
public Producer(SharedData data, int maxItems) {
this.data = data;
this.maxItems = maxItems;
}
@Override
public void run() {
for (int i = 1; i <= maxItems; i++) {
try {
data.produce(i);
Thread.sleep(500); // 模拟生产耗时
} catch (InterruptedException e) {
Thread.currentThread().in
terrupt();
break;
}
}
}
}
// 消费者线程
class Consumer implements Runnable {
private final SharedData data;
private final int maxItems;
public Consumer(SharedData data, int maxItems) {
this.data = data;
this.maxItems = maxItems;
}
@Override
public void run() {
for (int i = 0; i < maxItems; i++) {
try {
data.consume();
Thread.sleep(800); // 消费比生产慢
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}
}
// 主程序
public class ProducerConsumerDemo {
public static void main(String[] args) {
SharedData sharedData = new SharedData(5); // 队列容量为5
Thread producer = new Thread(new Producer(sharedData, 10));
Thread consumer = new Thread(new Consumer(sharedData, 10));
producer.start();
consumer.start();
try {
producer.join();
consumer.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("执行完成");
}
}
terrupt();
break;
}
}
}
}
// 消费者线程
class Consumer implements Runnable {
private final SharedData data;
private final int maxItems;
public Consumer(SharedData data, int maxItems) {
this.data = data;
this.maxItems = maxItems;
}
@Override
public void run() {
for (int i = 0; i < maxItems; i++) {
try {
data.consume();
Thread.sleep(800); // 消费比生产慢
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}
}
// 主程序
public class ProducerConsumerDemo {
public static void main(String[] args) {
SharedData sharedData = new SharedData(5); // 队列容量为5
Thread producer = new Thread(new Producer(sharedData, 10));
Thread consumer = new Thread(new Consumer(sharedData, 10));
producer.start();
consumer.start();
try {
producer.join();
consumer.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("执行完成");
}
}
3. 关键点说明
使用 BlockingQueue 实现生产者消费者的关键优势在于：
- 无需手动使用 synchronized、wait、notify 等机制
- put 和 take 方法自动处理阻塞逻辑
- 多生产者多消费者场景下天然支持，线程安全
- 可根据性能需求选择不同实现（如 LinkedBlockingQueue 通常吞吐更高）
注意在真实应用中，建议对中断做出响应，避免线程无法正常退出。
基本上就这些。BlockingQueue 让并发编程变得更简单可靠。