使用BlockingQueue实现生产者消费者模式可简化线程同步，通过put()和take()方法自动处理阻塞，无需手动控制锁；常用实现有ArrayBlockingQueue、LinkedBlockingQueue等；创建共享队列后，生产者添加任务，消费者取出处理，结合线程池可高效管理多线程协作，适用于高并发场景。
在Java中，使用BlockingQueue实现生产者消费者模式是一种高效且线程安全的方式。它简化了多线程编程中的同步问题，无需手动使用synchronized和wait/notify机制，底层已由队列自动处理阻塞与唤醒逻辑。
BlockingQueue简介
BlockingQueue是java.util.concurrent包下的接口，表示一个线程安全的队列，支持在插入或获取元素时进行阻塞操作：- 当队列满时，生产者线程调用
put()会被阻塞，直到有空间可用 - 当队列空时，消费者线程调用
take()会被阻塞，直到有元素可取
- ArrayBlockingQueue：基于数组的有界阻塞队列
- LinkedBlockingQueue：基于链表的可选有界队列
- PriorityBlockingQueue：支持优先级排序的无界阻塞队列
- SynchronousQueue：不存储元素的阻塞队列，每个插入必须等待对应移除
定义共享任务队列
创建一个BlockingQueue实例作为生产者和消费者之间的数据通道。例如，使用ArrayBlockingQueue限定最多存放10个任务：
BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);该队列将用于传递消息或任务对象，所有生产者向其中添加数据，所有消费者从中取出数据。
实现生产者线程
生产者模拟生成数据并放入队列。利用put()方法自动处理队列满时的阻塞：
class Producer implements Runnable {
private final BlockingQueue<String> queue;
public Producer(BlockingQueue<String> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 1; i <= 20; i++) {
String task = "Task-" + i;
queue.put(task);
System.out.println("Produced: " + task);
Thread.sleep(500); // 模拟生产耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
循环中每生成一个任务就调用put()，若队列满则自动暂停执行，直到消费者腾出空间。
实现消费者线程
消费者从队列中取出任务进行处理，使用take()方法在队列为空时自动阻塞：
class Consumer implements Runnable {
private final BlockingQueue<String> queue;
public Consumer(BlockingQueue<String> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (true) {
String task = queue.take();
System.out.println("Consumed: " + task);
Thread.sleep(800); // 模拟处理时间
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
消费者无限循环获取任务，take()会一直阻塞直到有新任务到达，适合长期运行的服务场景。
启动生产者和消费者线程
通过ExecutorService管理多个生产者和消费者线程：
public class ProducerConsumerDemo {
public static void main(String[] args) {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
ExecutorService executor = Executors.newFixedThreadPool(5);
executor.submit(new Producer(queue));
executor.submit(new Consumer(queue));
executor.submit(new Consumer(queue)); // 可启动多个消费者
// 运行一段时间后关闭
executor.shutdown();
try {
if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
}
}
}
此示例启动一个生产者和两个消费者，利用线程池统一调度资源。
基本上就这些。BlockingQueue让生产者消费者模式变得简洁可靠，关键是选择合适的实现类并合理设置容量，避免内存溢出或性能瓶颈。实际项目中可用于任务调度、消息中间件、日志处理等高并发场景。
