在Java多线程编程中，生产者消费者模式是非常经典的应用场景，多个生产者线程生成数据放入容器，多个消费者线程从容器中取出数据处理，需要实现线程间的协作与同步。传统方式需要手动使用synchronized配合wait、notify来实现阻塞和唤醒，容易出现逻辑漏洞。而BlockingQueue作为java.util.concurrent包下的接口，天生支持阻塞操作，非常适合实现这种场景。
BlockingQueue核心特性
BlockingQueue是Queue的子接口，具备以下核心特性：
- 线程安全：所有入队出队操作都是原子性的，不需要额外加锁
- 阻塞入队：当队列满时，put方法会阻塞当前线程直到队列有空位
- 阻塞出队：当队列空时，take方法会阻塞当前线程直到队列有新元素
- 提供了非阻塞和超时版本的入队出队方法，适配不同场景需求
实现阻塞生产者消费者步骤
1. 定义生产者线程
生产者线程负责生成数据，将数据通过put方法放入BlockingQueue，当队列满时会自动阻塞等待。
import java.util.concurrent.BlockingQueue;
public class Producer implements Runnable {
private final BlockingQueue<String> queue;
private final int produceCount; // 每个生产者生成的数据总量
public Producer(BlockingQueue<String> queue, int produceCount) {
this.queue = queue;
this.produceCount = produceCount;
}
@Override
public void run() {
try {
for (int i = 0; i < produceCount; i++) {
String data = "产品_" + Thread.currentThread().getName() + "_" + i;
// 队列满时会阻塞等待
queue.put(data);
System.out.println(Thread.currentThread().getName() + "生产了：" + data + "，当前队列大小：" + queue.size());
// 模拟生产耗时
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println(Thread.currentThread().getName() + "被中断");
}
}
}2. 定义消费者线程
消费者线程从BlockingQueue中通过take方法取数据，当队列空时会自动阻塞等待新数据到来。
import java.util.concurrent.BlockingQueue;
public class Consumer implements Runnable {
private final BlockingQueue<String> queue;
private final int consumeCount; // 每个消费者处理的数据总量
public Consumer(BlockingQueue<String> queue, int consumeCount) {
this.queue = queue;
this.consumeCount = consumeCount;
}
@Override
public void run() {
try {
for (int i = 0; i < consumeCount; i++) {
// 队列空时会阻塞等待
String data = queue.take();
System.out.println(Thread.currentThread().getName() + "消费了：" + data + "，当前队列大小：" + queue.size());
// 模拟消费耗时
Thread.sleep(150);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println(Thread.currentThread().getName() + "被中断");
}
}
}3. 编写测试主类
选择合适BlockingQueue实现类，启动生产者和消费者线程，验证阻塞效果。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class BlockingQueueDemo {
public static void main(String[] args) {
// 使用有界队列，容量为5，也可以根据需求选择LinkedBlockingQueue等实现
BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
ExecutorService executor = Executors.newFixedThreadPool(5);
// 启动2个生产者，每个生产10个产品
executor.submit(new Producer(queue, 10));
executor.submit(new Producer(queue, 10));
// 启动3个消费者，每个消费6个产品，总量和生产者匹配
executor.submit(new Consumer(queue, 6));
executor.submit(new Consumer(queue, 6));
executor.submit(new Consumer(queue, 8));
executor.shutdown();
}
}常用BlockingQueue实现类选择
根据不同的使用场景，可以选择不同的BlockingQueue实现：
| 实现类 | 特点 | 适用场景 |
|---|---|---|
| ArrayBlockingQueue | 基于数组实现的有界队列，构造时必须指定容量，内部使用单个ReentrantLock保证线程安全 | 需要严格控制队列大小，对性能要求较高的场景 |
| LinkedBlockingQueue | 基于链表实现，默认容量为Integer.MAX_VALUE，可指定容量，使用两把锁分别控制入队出队 | 吞吐量要求较高的场景，默认无界但建议指定容量避免内存溢出 |
| PriorityBlockingQueue | 支持优先级排序的无界队列，元素需要实现Comparable接口或构造时传入Comparator | 需要按优先级处理元素的场景 |
| SynchronousQueue | 不存储元素的队列，每个入队操作必须等待对应的出队操作，反之亦然 | 线程间直接传递数据的场景，吞吐量较高 |
注意事项
- 如果使用有界队列，要注意生产速度远大于消费速度时，生产者会频繁阻塞，需要合理设置队列容量
- 线程中断时，BlockingQueue的put和take方法会抛出InterruptedException，需要正确处理中断逻辑，避免数据丢失
- 不要混用BlockingQueue的阻塞方法和非阻塞方法，比如用add方法在满队列时插入会直接抛出异常，不符合阻塞生产者消费者的需求
- 如果需要实现优雅关闭，可以设置停止标志，让生产者不再生产新数据，消费者处理完剩余队列数据后再退出
BlockingQueue生产者消费者模式Java多线程线程阻塞修改时间：2026-05-25 21:25:32