使用阻塞队列可简化java中生产者消费者模式的实现,确保线程安全;也可通过synchronized与wait/notify或lock与condition实现更细粒度控制,关键在于正确处理共享资源的同步与线程通信。

在Java中实现线程安全的生产者消费者模式,核心是协调多个线程对共享资源的访问,确保数据一致性并避免死锁、竞态条件等问题。通常使用阻塞队列或结合synchronized与wait/notify机制来完成。
使用阻塞队列(推荐方式)
Java并发包java.util.concurrent提供了线程安全的阻塞队列,如ArrayBlockingQueue、LinkedBlockingQueue,天然支持生产者消费者模型。示例代码:
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
<p>class Producer implements Runnable {
private final BlockingQueue<Integer> queue;</p><pre class='brush:java;toolbar:false;'>public Producer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
public void run() {
try {
for (int i = 1; i <= 10; i++) {
queue.put(i); // 自动阻塞
System.out.println("生产者生产: " + i);
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}}
立即学习“Java免费学习笔记(深入)”;
class Consumer implements Runnable {
private final BlockingQueue
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
public void run() {
try {
while (true) {
Integer value = queue.take(); // 队列空时自动阻塞
System.out.println("消费者消费: " + value);
Thread.sleep(200);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}}
立即学习“Java免费学习笔记(深入)”;
public class ProducerConsumerExample {
public static void main(String[] args) {
BlockingQueue
Thread producer = new Thread(new Producer(queue));
Thread consumer = new Thread(new Consumer(queue));
producer.start();
consumer.start();
}}
立即学习“Java免费学习笔记(深入)”;
优点:无需手动管理同步,put和take方法自动处理阻塞与唤醒,代码简洁且不易出错。
使用synchronized + wait/notify
当需要自定义缓冲区结构时,可以使用synchronized关键字保护共享资源,并通过wait和notifyAll控制线程协作。关键点:
- 共享资源(如List或数组)必须被synchronized保护
- 条件判断用while而不是if,防止虚假唤醒
- 调用wait会释放锁,notify唤醒等待线程
示例代码片段:
class SharedBuffer {
private final int MAX_SIZE = 5;
private List<Integer> buffer = new ArrayList<>();
<pre class='brush:java;toolbar:false;'>public void produce(int item) throws InterruptedException {
synchronized (this) {
while (buffer.size() == MAX_SIZE) {
this.wait(); // 缓冲区满,生产者等待
}
buffer.add(item);
System.out.println("生产: " + item);
this.notifyAll(); // 唤醒消费者
}
}
public int consume() throws InterruptedException {
synchronized (this) {
while (buffer.isEmpty()) {
this.wait(); // 缓冲区空,消费者等待
}
int item = buffer.remove(buffer.size() - 1);
System.out.println("消费: " + item);
this.notifyAll(); // 唤醒生产者
return item;
}
}}
立即学习“Java免费学习笔记(深入)”;
注意:必须使用while循环检查条件,否则可能因虚假唤醒导致逻辑错误。
使用Lock和Condition(更灵活的控制)
ReentrantLock配合Condition接口可实现更细粒度的线程通信,比如分别定义“非满”和“非空”两个条件。优势:
- 支持多个等待集(多个Condition)
- 可中断等待(lockInterruptibly)
- 公平锁选项
典型用法:
import java.util.concurrent.locks.*;
<p>class BufferWithCondition {
private final int[] buffer = new int[5];
private int count = 0, in = 0, out = 0;
private final Lock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();</p><pre class='brush:java;toolbar:false;'>public void put(int item) throws InterruptedException {
lock.lock();
try {
while (count == buffer.length) {
notFull.await(); // 等待非满
}
buffer[in] = item;
in = (in + 1) % buffer.length;
count++;
notEmpty.signal(); // 通知消费者
} finally {
lock.unlock();
}
}
public int take() throws InterruptedException {
lock.lock();
try {
while (count == 0) {
notEmpty.await(); // 等待非空
}
int item = buffer[out];
out = (out + 1) % buffer.length;
count--;
notFull.signal(); // 通知生产者
return item;
} finally {
lock.unlock();
}
}}
立即学习“Java免费学习笔记(深入)”;
这种方式比synchronized更灵活,适合复杂场景。
基本上就这些。选择哪种方式取决于需求:简单场景用阻塞队列最安全高效;需要定制逻辑可用synchronized+wait/notify或Lock+Condition。关键是保证共享状态的可见性与原子性,合理使用等待唤醒机制避免忙等。










