BlockingQueue是Java中实现生产者消费者模式的理想工具，通过put()和take()方法实现线程安全的阻塞操作，无需手动同步。使用ArrayBlockingQueue等实现类创建有界队列，生产者调用put()插入数据，消费者调用take()获取数据，自动处理等待与通知。通过Thread或ExecutorService启动生产者消费者线程，合理设置队列容量并处理中断，即可高效稳定运行系统。
在Java中，BlockingQueue 是实现生产者消费者模式的理想工具。它是一个线程安全的队列，支持阻塞的插入和移除操作，能自动处理多线程环境下的同步问题，无需手动加锁。下面详细介绍如何使用 BlockingQueue 实现生产者消费者模型。
1. BlockingQueue 简介与核心方法
BlockingQueue 是 java.util.concurrent 包中的接口，常见实现类有 ArrayBlockingQueue（基于数组、有界）、LinkedBlockingQueue（基于链表、可设界）、SynchronousQueue（不存储元素）等。
关键方法包括：
- put(E e)：将元素插入队列，如果队列满，则阻塞等待空间
- take()：从队列获取元素，如果队列空，则阻塞等待元素
- offer(E e, time, unit) 和 poll(time, unit)：带超时的插入和取出，非永久阻塞
2. 定义共享的 BlockingQueue
生产者和消费者通过同一个 BlockingQueue 实例通信。通常将其声明为成员变量或通过构造函数传入。
立即学习“Java免费学习笔记（深入）”；
示例代码：
BlockingQueuequeue = new ArrayBlockingQueue<>(10); // 容量为10
3. 实现生产者线程
生产者不断生成数据并放入队列。put() 方法会在队列满时自动阻塞。
class Producer implements Runnable {
private final BlockingQueue queue;
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
for (int i = 1; i <= 20; i++) {
String task = "任务-" + i;
queue.put(task);
System.out.println("生产者: 生产了 " + task);
Thread.sleep(500); // 模拟生产耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
4. 实现消费者线程
消费者从队列中取出数据处理。take() 方法在队列为空时会自动等待。
class Consumer implements Runnable {
private final BlockingQueue queue;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
while (true) {
String task = queue.take();
System.out.println("消费者: 消费了 " + task);
Thread.sleep(800); // 模拟消费耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
5. 启动生产者和消费者线程
使用 Thread 或线程池启动多个生产者和消费者。
public class ProducerConsumerDemo {
public static void main(String[] args) {
BlockingQueue queue = new ArrayBlockingQueue<>(5);
Thread producer = new Thread(new Producer(queue));
Thread consumer = new Thread(new Consumer(queue));
producer.start();
consumer.start();
try {
producer.join();
consumer.interrupt(); // 生产结束，中断消费者
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
注意：实际应用中可用 volatile 标志位控制循环退出，避免无限消费。
6. 使用线程池优化资源管理
更推荐使用 ExecutorService 管理线程生命周期。
ExecutorService executor = Executors.newFixedThreadPool(3); executor.submit(new Producer(queue)); executor.submit(new Consumer(queue)); // 提交多个消费者或生产者 executor.shutdown();
基本上就这些。BlockingQueue 封装了复杂的同步逻辑，让开发者专注业务实现。只要合理设置队列容量，正确处理中断，就能高效稳定地运行生产者消费者系统。
