多线程协作中的生产者与消费者模式实现

10 0 0

概述

在多线程编程领域，生产者-消费者架构是一种广泛使用的并发协作范式。深入理解这一模式有助于开发者掌握线程同步的核心机制，这也是技术面试中经常考察的编程实践题目。
该模式主要涉及两种执行线程：负责生成数据的生产者线程和负责处理数据的消费者线程。为了实现两者的解耦，通常会设置一个共享存储区域作为中介。生产者只需将数据存入该区域，无需关注消费过程；消费者则直接从该区域获取数据，不必了解生产细节。
这个共享存储区域需要满足以下同步要求：
1. 当存储区达到容量上限时，暂停生产者的数据生成操作
2. 当存储区为空时，暂停消费者的数据获取操作
常见的实现方案包括以下三种：
1. 基于阻塞队列的实现方式
2. 采用synchronized配合wait/notify机制
3. 使用Lock配合Condition的await/signal机制

基于阻塞队列的实现方案

阻塞队列内置了线程安全的插入和移除方法。当队列满时，生产者线程会自动阻塞；当队列空时，消费者线程会自动挂起。这种机制让开发者可以专注于业务逻辑的实现。

public class ProducerConsumerDemo {
private static LinkedBlockingQueue sharedQueue = new LinkedBlockingQueue<>();
public static void main(String[] args) {
ExecutorService pool = Executors.newFixedThreadPool(15);
for (int i = 0; i < 5; i++) {
pool.execute(new Producer());
pool.execute(new Consumer());
}
}
}

使用synchronized的实现方式

通过wait/notify机制可以手动控制线程的阻塞和唤醒。以下示例展示了一个自定义的线程安全队列：

public class CustomBlockingQueue {
private Queue<String> dataQueue = new LinkedList<>();
private final int CAPACITY = 20;
public synchronized void produce(String item) throws Exception {
while (dataQueue.size() == CAPACITY) {
wait();
}
dataQueue.offer(item);
notifyAll();
}
public synchronized String consume() throws Exception {
while (dataQueue.isEmpty()) {
wait();
}
String item = dataQueue.poll();
notifyAll();
return item;
}
}

基于Lock和Condition的实现

这种方案提供了更灵活的线程控制方式，可以精确指定唤醒的线程类型：

public class BoundedBuffer {
private final Queue<Object> buffer;
private final int capacity;
private final Lock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
public void put(Object obj) throws InterruptedException {
lock.lock();
try {
while (buffer.size() == capacity) {
notFull.await();
}
buffer.add(obj);
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (buffer.isEmpty()) {
notEmpty.await();
}
Object obj = buffer.poll();
notFull.signal();
return obj;
} finally {
lock.unlock();
}
}
}