使用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该队列将用于传递消息或任务对象,所有生产者向其中添加数据,所有消费者从中取出数据。queue = new ArrayBlockingQueue<>(10);
实现生产者线程
生产者模拟生成数据并放入队列。利用put()方法自动处理队列满时的阻塞:
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 <= 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 queue;
public Consumer(BlockingQueue 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 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让生产者消费者模式变得简洁可靠,关键是选择合适的实现类并合理设置容量,避免内存溢出或性能瓶颈。实际项目中可用于任务调度、消息中间件、日志处理等高并发场景。










