生产者消费者问题是并发编程的经典模型。多个生产者线程生成数据放入缓冲区，多个消费者线程从缓冲区取出数据消费。核心挑战是：缓冲区满时生产者需等待，缓冲区空时消费者需等待，且必须保证线程安全。
问题建模
- 共享缓冲区容量固定为
N - 生产者：生产物品，若缓冲区满则阻塞等待
- 消费者：消费物品，若缓冲区空则阻塞等待
- 同步需求：互斥访问缓冲区 + 条件等待
核心算法：信号量同步
使用 Semaphore 实现精确控制：
– emptySlots：空槽位信号量，初始值为 N
– filledSlots：已填充槽位信号量，初始值为 0
– mutex：互斥锁，保证缓冲区操作原子性
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生产者消费者问题求解器
* 使用信号量实现多线程同步与互斥
*/
public class ProducerConsumer {
/**
* 共享缓冲区，使用Queue实现FIFO
*/
static class Buffer<T> {
private final Queue<T> queue;
private final int capacity;
// 信号量控制
private final Semaphore emptySlots; // 空槽位数量
private final Semaphore filledSlots; // 已填充槽位数量
private final Semaphore mutex; // 互斥锁（二进制信号量）
public Buffer(int capacity) {
this.capacity = capacity;
this.queue = new LinkedList<>();
this.emptySlots = new Semaphore(capacity); // 初始有capacity个空位
this.filledSlots = new Semaphore(0); // 初始没有物品
this.mutex = new Semaphore(1); // 互斥访问
}
/**
* 生产者放入物品
*/
public void produce(T item) throws InterruptedException {
// 申请一个空槽位，如果没有则阻塞等待
emptySlots.acquire();
// 获取互斥锁，进入临界区
mutex.acquire();
try {
queue.offer(item);
System.out.println(Thread.currentThread().getName() + " 生产: " + item
+ " | 缓冲区大小: " + queue.size());
} finally {
mutex.release();
}
// 释放一个已填充槽位信号量，通知消费者
filledSlots.release();
}
/**
* 消费者取出物品
*/
public T consume() throws InterruptedException {
// 申请一个已填充槽位，如果没有则阻塞等待
filledSlots.acquire();
// 获取互斥锁，进入临界区
mutex.acquire();
T item;
try {
item = queue.poll();
System.out.println(Thread.currentThread().getName() + " 消费: " + item
+ " | 缓冲区大小: " + queue.size());
} finally {
mutex.release();
}
// 释放一个空槽位信号量，通知生产者
emptySlots.release();
return item;
}
}
static class Producer implements Runnable {
private final Buffer<Integer> buffer;
private final AtomicInteger counter;
private final int produceCount;
Producer(Buffer<Integer> buffer, AtomicInteger counter, int produceCount) {
this.buffer = buffer;
this.counter = counter;
this.produceCount = produceCount;
}
@Override
public void run() {
try {
for (int i = 0; i < produceCount; i++) {
int item = counter.incrementAndGet();
buffer.produce(item);
Thread.sleep((int)(Math.random() * 100)); // 模拟生产耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
static class Consumer implements Runnable {
private final Buffer<Integer> buffer;
private final int consumeCount;
Consumer(Buffer<Integer> buffer, int consumeCount) {
this.buffer = buffer;
this.consumeCount = consumeCount;
}
@Override
public void run() {
try {
for (int i = 0; i < consumeCount; i++) {
buffer.consume();
Thread.sleep((int)(Math.random() * 150)); // 模拟消费耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
public static void main(String[] args) throws InterruptedException {
final int BUFFER_SIZE = 5;
final int PRODUCER_COUNT = 3;
final int CONSUMER_COUNT = 2;
final int ITEMS_PER_PRODUCER = 4;
final int ITEMS_PER_CONSUMER = 6;
Buffer<Integer> buffer = new Buffer<>(BUFFER_SIZE);
AtomicInteger counter = new AtomicInteger(0);
Thread[] producers = new Thread[PRODUCER_COUNT];
Thread[] consumers = new Thread[CONSUMER_COUNT];
// 启动生产者线程
for (int i = 0; i < PRODUCER_COUNT; i++) {
producers[i] = new Thread(new Producer(buffer, counter, ITEMS_PER_PRODUCER),
"Producer-" + i);
producers[i].start();
}
// 启动消费者线程
for (int i = 0; i < CONSUMER_COUNT; i++) {
consumers[i] = new Thread(new Consumer(buffer, ITEMS_PER_CONSUMER),
"Consumer-" + i);
consumers[i].start();
}
// 等待所有线程完成
for (Thread t : producers) t.join();
for (Thread t : consumers) t.join();
System.out.println("所有生产消费任务完成！");
}
}
另一种实现：BlockingQueue（更简洁）
import java.util.concurrent.*;
public class ProducerConsumerBlockingQueue {
public static void main(String[] args) throws InterruptedException {
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
AtomicInteger counter = new AtomicInteger(0);
// 生产者
Runnable producer = () -> {
try {
for (int i = 0; i < 5; i++) {
int item = counter.incrementAndGet();
queue.put(item); // 队列满时自动阻塞
System.out.println(Thread.currentThread().getName() + " 生产: " + item);
Thread.sleep(50);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
};
// 消费者
Runnable consumer = () -> {
try {
for (int i = 0; i < 7; i++) {
Integer item = queue.take(); // 队列空时自动阻塞
System.out.println(Thread.currentThread().getName() + " 消费: " + item);
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
};
new Thread(producer, "P1").start();
new Thread(producer, "P2").start();
new Thread(consumer, "C1").start();
new Thread(consumer, "C2").start();
}
}
复杂度分析
- 时间复杂度：单次
produce或consume操作均为O(1)（忽略线程调度开销）。 - 空间复杂度：
O(N)，N为缓冲区容量。 - 线程安全：信号量保证了互斥与同步，不存在竞态条件或死锁（获取顺序固定：
emptySlots/filledSlots->mutex）。
常见变体与扩展
- 多缓冲区：使用多个缓冲区减少竞争，提升并发度。
- 优先级队列：按优先级消费，使用
PriorityBlockingQueue。 - 限时等待：使用
Semaphore.tryAcquire(timeout)实现非阻塞超时报错。 - 读者写者问题：将生产者和消费者扩展为读者和写者，允许多个读者并发。
总结
生产者消费者问题展示了并发编程的核心挑战：资源共享与线程协作。通过信号量精确控制缓冲区状态，既能保证线程安全，又能实现高效的同步等待机制。
