生产者-消费者模型是多线程问题里面的经典问题，也是面试的常见问题。有如下几个常见的实现方法：
1. wait()/notify()
2. lock & condition
3. BlockingQueue
下面来逐一分析。
1. wait()/notify()
第一种实现，利用根类Object的两个方法wait()/notify()，来停止或者唤醒线程的执行；这也是最原始的实现。
public class WaitNotifyBroker<T> implements Broker<T> {
private final Object[] items;
private int takeIndex;
private int putIndex;
private int count;
public WaitNotifyBroker(int capacity) {
this.items = new Object[capacity];
}
@SuppressWarnings("unchecked")
@Override
public T take() {
T tmpObj = null;
try {
synchronized (items) {
while (0 == count) {
items.wait();
}
tmpObj = (T) items[takeIndex];
if (++takeIndex == items.length) {
takeIndex = 0;
}
count--;
items.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
return tmpObj;
}
@Override
public void put(T obj) {
try {
synchronized (items) {
while (items.length == count) {
items.wait();
}
items[putIndex] = obj;
if (++putIndex == items.length) {
putIndex = 0;
}
count++;
items.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
这里利用Array构造一个Buffer去存取数据，并利用count, putIndex和takeIndex来保证First-In-First-Out。
如果利用LinkedList来代替Array，相对来说会稍微简单些。
LinkedList的实现，可以参考《Java 7 Concurrency Cookbook》第2章wait/notify。
2. lock & condition
lock & condition，实际上也实现了类似synchronized和wait()/notify()的功能，但在加锁和解锁、暂停和唤醒方面，更加细腻和可控。
在JDK的BlockingQueue的默认实现里，也是利用了lock & condition。此文也详细介绍了怎么利用lock&condition写BlockingQueue，这里换LinkedList再实现一次：
public class LockConditionBroker<T> implements Broker<T> {
private final ReentrantLock lock;
private final Condition notFull;
private final Condition notEmpty;
private final int capacity;
private LinkedList<T> items;
public LockConditionBroker(int capacity) {
this.lock = new ReentrantLock();
this.notFull = lock.newCondition();
this.notEmpty = lock.newCondition();
this.capacity = capacity;
items = new LinkedList<T>();
}
@Override
public T take() {
T tmpObj = null;
lock.lock();
try {
while (items.size() == 0) {
notEmpty.await();
}
tmpObj = items.poll();
notFull.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return tmpObj;
}
@Override
public void put(T obj) {
lock.lock();
try {
while (items.size() == capacity) {
notFull.await();
}
items.offer(obj);
notEmpty.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
3. BlockingQueue
最后这种方法，也是最简单最值得推荐的。利用并发包提供的工具：阻塞队列，将阻塞的逻辑交给BlockingQueue。
实际上，上述1和2的方法实现的Broker类，也可以视为一种简单的阻塞队列，不过没有标准包那么完善。
public class BlockingQueueBroker<T> implements Broker<T> {
private final BlockingQueue<T> queue;
public BlockingQueueBroker() {
this.queue = new LinkedBlockingQueue<T>();
}
@Override
public T take() {
try {
return queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
@Override
public void put(T obj) {
try {
queue.put(obj);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
我们的队列封装了标注包里的LinkedBlockingQueue，十分简单高效。
接下来，就是一个1P2C的例子：
public interface Broker<T> {
T take();
void put(T obj);
}
public class Producer implements Runnable {
private final Broker<Integer> broker;
private final String name;
public Producer(Broker<Integer> broker, String name) {
this.broker = broker;
this.name = name;
}
@Override
public void run() {
try {
for (int i = 0; i < 5; i++) {
broker.put(i);
System.out.format("%s produced: %s%n", name, i);
Thread.sleep(1000);
}
broker.put(-1);
System.out.println("produced termination signal");
} catch (InterruptedException e) {
e.printStackTrace();
return;
}
}
}
public class Consumer implements Runnable {
private final Broker<Integer> broker;
private final String name;
public Consumer(Broker<Integer> broker, String name) {
this.broker = broker;
this.name = name;
}
@Override
public void run() {
try {
for (Integer message = broker.take(); message != -1; message = broker.take()) {
System.out.format("%s consumed: %s%n", name, message);
Thread.sleep(1000);
}
System.out.println("received termination signal");
} catch (InterruptedException e) {
e.printStackTrace();
return;
}
}
}
public class Main {
public static void main(String[] args) {
Broker<Integer> broker = new WaitNotifyBroker<Integer>(5);
// Broker<Integer> broker = new LockConditionBroker<Integer>(5);
// Broker<Integer> broker = new BlockingQueueBroker<Integer>();
new Thread(new Producer(broker, "prod 1")).start();
new Thread(new Consumer(broker, "cons 1")).start();
new Thread(new Consumer(broker, "cons 2")).start();
}
}
除了上述的方法，其实还有很多第三方的并发包可以解决这个问题。例如LMAX Disruptor和Chronicle等
本文完。
参考：
《Java 7 Concurrency Cookbook》