生产者-消费者模型是多线程问题里面的经典问题，也是面试的常见问题。有如下几个常见的实现方法：

1. wait()/notify()

2. lock & condition

3. BlockingQueue

下面来逐一分析。

1. wait()/notify()

第一种实现，利用根类Object的两个方法wait()/notify()，来停止或者唤醒线程的执行；这也是最原始的实现。

1 public class WaitNotifyBroker implements Broker{2

3 private finalObject[] items;4

5 private inttakeIndex;6 private intputIndex;7 private intcount;8

9 public WaitNotifyBroker(intcapacity) {10 this.items = newObject[capacity];11 }12

13 @SuppressWarnings("unchecked")14 @Override15 publicT take() {16 T tmpObj = null;17 try{18 synchronized(items) {19 while (0 ==count) {20 items.wait();21 }22 tmpObj =(T) items[takeIndex];23 if (++takeIndex ==items.length) {24 takeIndex = 0;25 }26 count--;27 items.notify();28 }29 } catch(InterruptedException e) {30 e.printStackTrace();31 }32

33 returntmpObj;34 }35

36 @Override37 public voidput(T obj) {38 try{39 synchronized(items) {40 while (items.length ==count) {41 items.wait();42 }43

44 items[putIndex] =obj;45 if (++putIndex ==items.length) {46 putIndex = 0;47 }48 count++;49 items.notify();50 }51 } catch(InterruptedException e) {52 e.printStackTrace();53 }54

55 }56

57 }

这里利用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再实现一次：

1 public class LockConditionBroker implements Broker{2

3 private finalReentrantLock lock;4 private finalCondition notFull;5 private finalCondition notEmpty;6 private final intcapacity;7 private LinkedListitems;8

9 public LockConditionBroker(intcapacity) {10 this.lock = newReentrantLock();11 this.notFull =lock.newCondition();12 this.notEmpty =lock.newCondition();13 this.capacity =capacity;14

15 items = new LinkedList();16 }17

18 @Override19 publicT take() {20 T tmpObj = null;21 lock.lock();22 try{23 while (items.size() == 0) {24 notEmpty.await();25 }26

27 tmpObj =items.poll();28 notFull.signalAll();29

30 } catch(InterruptedException e) {31 e.printStackTrace();32 } finally{33 lock.unlock();34 }35 returntmpObj;36 }37

38 @Override39 public voidput(T obj) {40 lock.lock();41 try{42 while (items.size() ==capacity) {43 notFull.await();44 }45

46 items.offer(obj);47 notEmpty.signalAll();48

49 } catch(InterruptedException e) {50 e.printStackTrace();51 } finally{52 lock.unlock();53 }54

55 }56 }

3. BlockingQueue

最后这种方法，也是最简单最值得推荐的。利用并发包提供的工具：阻塞队列，将阻塞的逻辑交给BlockingQueue。

实际上，上述1和2的方法实现的Broker类，也可以视为一种简单的阻塞队列，不过没有标准包那么完善。

1 public class BlockingQueueBroker implements Broker{2

3 private final BlockingQueuequeue;4

5 publicBlockingQueueBroker() {6 this.queue = new LinkedBlockingQueue();7 }8

9 @Override10 publicT take() {11 try{12 returnqueue.take();13 } catch(InterruptedException e) {14 e.printStackTrace();15 }16

17 return null;18 }19

20 @Override21 public voidput(T obj) {22 try{23 queue.put(obj);24 } catch(InterruptedException e) {25 e.printStackTrace();26 }27 }28

29 }

我们的队列封装了标注包里的LinkedBlockingQueue，十分简单高效。

接下来，就是一个1P2C的例子：

1 public interface Broker{2

3 T take();4

5 voidput(T obj);6

7 }8

9

10 public class Producer implementsRunnable {11

12 private final Brokerbroker;13 private finalString name;14

15 public Producer(Brokerbroker, String name) {16 this.broker =broker;17 this.name =name;18 }19

20 @Override21 public voidrun() {22 try{23 for (int i = 0; i < 5; i++) {24 broker.put(i);25 System.out.format("%s produced: %s%n", name, i);26 Thread.sleep(1000);27 }28 broker.put(-1);29 System.out.println("produced termination signal");30 } catch(InterruptedException e) {31 e.printStackTrace();32 return;33 }34

35 }36

37 }38

39

40 public class Consumer implementsRunnable {41

42 private final Brokerbroker;43 private finalString name;44

45 public Consumer(Brokerbroker, String name) {46 this.broker =broker;47 this.name =name;48 }49

50 @Override51 public voidrun() {52 try{53 for (Integer message = broker.take(); message != -1; message =broker.take()) {54 System.out.format("%s consumed: %s%n", name, message);55 Thread.sleep(1000);56 }57 System.out.println("received termination signal");58 } catch(InterruptedException e) {59 e.printStackTrace();60 return;61 }62

63 }64

65 }66

67

68 public classMain {69

70 public static voidmain(String[] args) {71 Broker broker = new WaitNotifyBroker(5);72 //Broker broker = new LockConditionBroker(5);73 //Broker broker = new BlockingQueueBroker();

74

75 new Thread(new Producer(broker, "prod 1")).start();76 new Thread(new Consumer(broker, "cons 1")).start();77 new Thread(new Consumer(broker, "cons 2")).start();78

79 }80

81 }

除了上述的方法，其实还有很多第三方的并发包可以解决这个问题。例如LMAX Disruptor和Chronicle等

本文完。

参考：

《Java 7 Concurrency Cookbook》