使用BlockingQueue可简化线程安全的生产者消费者模型，其内置阻塞机制能自动处理队列满或空的情况，避免手动加锁；结合synchronized与wait/notify或ReentrantLock与Condition可实现更细粒度控制，适用于需自定义同步逻辑的场景。
在Java中实现线程安全的生产者消费者模型，核心在于协调多个线程对共享资源的访问，避免数据竞争和状态不一致。通常使用阻塞队列、锁机制或信号量来控制生产者和消费者的协作。下面从实践角度出发，介绍几种常见且可靠的实现方式。
使用BlockingQueue实现线程安全
Java提供了java.util.concurrent.BlockingQueue接口及其实现类（如ArrayBlockingQueue、LinkedBlockingQueue），天然支持线程安全的生产者消费者模式。
BlockingQueue的关键特性是：当队列满时，生产者线程自动阻塞；当队列空时，消费者线程自动阻塞，直到有新元素可用。
示例代码：
立即学习“Java免费学习笔记（深入）”；
import java.util.concurrent.*;
public class ProducerConsumerWithQueue {
private BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
class Producer implements Runnable {
public void run() {
try {
for (int i = 1; i <= 10; i++) {
String item = "Item-" + i;
queue.put(item); // 自动阻塞
System.out.println("生产者生产: " + item);
Thread.sleep(500);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
class Consumer implements Runnable {
public void run() {
try {
while (true) {
String item = queue.take(); // 自动阻塞
System.out.println("消费者消费: " + item);
Thread.sleep(800);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
public static void main(String[] args) {
ProducerConsumerWithQueue pc = new ProducerConsumerWithQueue();
new Thread(pc.new Producer()).start();
new Thread(pc.new Consumer()).start();
}
}
这种实现无需手动加锁，代码简洁，推荐在大多数场景下优先使用。
使用synchronized与wait/notify机制
在没有阻塞队列的情况下，可以基于synchronized关键字配合wait()和notifyAll()方法手动实现线程通信。
需要一个共享的缓冲区，并确保所有操作都在同步块中进行。
关键点：
- 使用while循环检查条件，防止虚假唤醒
- 调用wait()会释放锁，notifyAll()唤醒所有等待线程
- 生产者和消费者都需获取同一对象的锁
示例片段：
private final List<String> buffer = new ArrayList<>();
private final int MAX_SIZE = 5;
public void produce(String item) throws InterruptedException {
synchronized (buffer) {
while (buffer.size() == MAX_SIZE) {
buffer.wait(); // 等待空间
}
buffer.add(item);
System.out.println("生产: " + item);
buffer.notifyAll(); // 唤醒消费者
}
}
public String consume() throws InterruptedException {
synchronized (buffer) {
while (buffer.isEmpty()) {
buffer.wait(); // 等待数据
}
String item = buffer.remove(0);
System.out.println("消费: " + item);
buffer.notifyAll(); // 唤醒生产者
return item;
}
}
这种方式更底层，适合理解原理，但容易出错，需谨慎处理异常和中断。
使用ReentrantLock与Condition
相比synchronized，ReentrantLock提供更灵活的控制能力，结合Condition可分别定义“非满”和“非空”两个等待条件。
优势：
- 可以创建多个Condition实例，实现精准唤醒
- 支持尝试获取锁、超时等高级操作
- 性能在高并发下通常优于synchronized
示例：
private final Lock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
private final List<String> buffer = new ArrayList<>();
private final int MAX_SIZE = 5;
public void produce(String item) throws InterruptedException {
lock.lock();
try {
while (buffer.size() == MAX_SIZE) {
notFull.await(); // 等待不满
}
buffer.add(item);
System.out.println("生产: " + item);
notEmpty.signal(); // 唤醒消费者
} finally {
lock.unlock();
}
}
public String consume() throws InterruptedException {
lock.lock();
try {
while (buffer.isEmpty()) {
notEmpty.await(); // 等待不空
}
String item = buffer.remove(0);
System.out.println("消费: " + item);
notFull.signal(); // 唤醒生产者
return item;
} finally {
lock.unlock();
}
}
这种写法比wait/notify更清晰，尤其适合复杂同步逻辑。
实际应用中的注意事项
在真实项目中，除了基本模型，还需关注以下几点：
- 合理设置缓冲区大小，避免内存溢出或频繁阻塞
- 处理线程中断，确保资源正确释放
- 避免死锁，注意锁的顺序和持有时间
- 考虑使用线程池管理生产者和消费者线程
- 监控队列长度和处理延迟，便于性能调优
对于高吞吐场景，可选择LinkedBlockingQueue或Disruptor等高性能队列。
基本上就这些。选择哪种方式取决于具体需求：BlockingQueue最简单可靠，适合大多数情况；自定义同步机制则用于学习或特殊控制场景。关键是保证线程安全和响应性，同时代码易于维护。
