上面说了一大堆故事。下面就开始今天的主题： 生产者消费者模型。这里先问两个问题：1、 生产者消费者模型旨在解决什么问题。2、 在现有的场景下哪里会用到？

从下面图中可以发现生产者和消费者之间用中间类似一个队列一样的东西串起来。这个队列可以想像成一个存放产品的“ 仓库”，生产者只需要关心这个“ 仓库”，并不需要关心具体的消费者，对于生产者而言甚至都不知道有这些消费者存在。对于消费者而言他也不需要关心 具体的生产者，到底有多少生产者也不是他关心的事情，他只要关心这个“ 仓库”中还有没有东西。这种模型是一种松耦合模型。这样可以回答我上面提出的第一个问题。这个模型的产生就是为了 复用和解耦。比如常见的消息框架（非常经典的一种生产者消费者模型的使用场景） ActiveMQ。发送端和接收端用Topic进行关联。这个Topic可以理解为我们这里“ 仓库”的地址，这样就可以实现点对点和广播两种方式进行消息的分发。

       我的代码里只是简单的模拟了这样一个模型的实现。在写的过程中还是碰到一些问题。编程过程中一旦用多线程复杂度就会陡升。所以编码原则是 能用单线程解决问题就不要去用多线程。

图中比较简单的画了生产者和消费者之间的关系，如果看到这里还比较迷惑没有关系，下面会有具体的代码展示。当然最好的方式就是自己从头开始写，这样可以知道一些在看文章中遗漏的一些细节。在学习的过程中往往会有这样的感受： 你以为你看懂了，理解了，自己写的时候就发现傻眼了，这也是为什么代码一定要多写，写的过程是即是思考的过程。

下面我会重点讲两个类的代码，Consumer.java和Producer.java，以及我在写的过程中碰到了些什么问题。现在有点发现用一篇的篇幅有点讲不清楚，代码我会在后面贴出来，我估计在这里贴出来也很少人会去看这个代码。从我自己的经验来看一看到一大坨的代码就傻眼了。

先回答我上面的两把锁的问题，对于生产者和消费者的唤醒和挂起操作分别用了两个监控器。原因很简单，就像我在生产者消费者模型图里面看到的生产者和消费者均可以是一个或者多个。如果这里用一个对象控制，在多个生产者和多个消费者的时候发现：本来要唤醒生产者会把消费者也唤醒，就会比较混乱。

再看第二个问题是： 为什么在三个地方都需要做判满操作？这里看生产者那张图，仔细的看这一段代码，这里的操作 并非原子操作。图中标识的1处判满操作后，到了第3处后并不意味着容器还是满的。如果只是在1处作了校验，那么很有可能会出现这样的情况：在2处把消费者唤醒后，等运行到了第3处发现 这时候的容器可能已经空了！这样就不需要把生产者进行挂起。而如果没有这段校验逻辑很有可能就会出现 生产者和消费者同时挂起的情况，这里可以细细体会一下，或者写代码试验一下。

         这里再说一下wait()和notify()/notifyAll()。这两个操作是一对。这里要注意的第一个问题是对于线程进行wait()和notify()操作都要 锁住持有对象。图中都可以看出代码中对两个monitor都加了锁，不加锁的话就会抛出 java.lang.IllegalMonitorStateException异常。另外一个问题很容易的一个惯性思维就是线程挂起不就是this.wait()，这样可以吗？答案是可以，你只要锁住自己就OK。但是问题来了， 谁来唤醒它？这也是为什么用一个公用对象进行唤醒和挂起线程操作，wait()和notify()属于Object类，任何对象都可以进行这个操作。

/*** 生产者* @author 百恼  2013-11-7下午12:18:39**/
public class Producer implements Runnable {//简单的模拟，这里一个生产容器，设置成final类型的话不允许再次赋值private final Container<Bread> container;//生产者线程监听器private final Object producerMonitor;//消费者线程监听器private final Object consumerMonitor;public Producer(Object producerMonitor,Object consumerMonitor,Container<Bread> container){this.producerMonitor = producerMonitor;this.consumerMonitor = consumerMonitor;this.container = container;}/* (non-Javadoc)* @see java.lang.Runnable#run()*/@Overridepublic void run() {while(true){produce();}}public void produce(){//这里为了形象，模拟几个制作面包的步骤step1();Bread bread = step2();//如果发现容器已经满了,生产者要停if(container.isFull()){//唤醒消费者synchronized(consumerMonitor){if(container.isFull()){consumerMonitor.notify();}}//生产者挂起,两把锁的问题synchronized(producerMonitor){try {if(container.isFull()){System.out.println("生产者挂起...");producerMonitor.wait();                        }} catch (InterruptedException e) {e.printStackTrace();}}}else{//容器中还有容量,把面包放到容器内,这里可能会有丢失boolean result = container.add(bread);System.out.println("Producer:"+result);}}public void step1(){}public Bread step2(){return new Bread();}
}

Consumer.java

/*** 消费者* @author 百恼  2013-11-7下午12:18:50**/
public class Consumer implements Runnable{//简单的模拟，这里一个生产容器，设置成final类型的话不允许再次赋值private final Container<Bread> container;//生产者线程监听器private final Object producerMonitor;//消费者线程监听器private final Object consumerMonitor;public Consumer(Object producerMonitor,Object consumerMonitor,Container<Bread> container){this.producerMonitor = producerMonitor;this.consumerMonitor = consumerMonitor;this.container = container;}@Overridepublic void run() {while(true){consume();}}//消费，两把锁的问题public void consume(){//如果发现容器已经满了,生产者要停if(container.isEmpty()){//唤醒生产者synchronized(producerMonitor){if(container.isEmpty()){producerMonitor.notify();}}                //消费者挂起synchronized(consumerMonitor){try {if(container.isEmpty()){System.out.println("消费者挂起。。。");consumerMonitor.wait();                        }} catch (InterruptedException e) {e.printStackTrace();}}}else{//还有面包可以进行消费Bread bread = container.get();System.out.println("bread:"+bread);}       }
}

Container.java

/*** 装产品的容器* @author 百恼 2013-11-9下午02:06:59**/
public class Container<T> {private final int capacity;private final List<T> list;public Container(int capacity){this.capacity = capacity;list = new ArrayList<T>(capacity);}public List<T> getList(){return list;}/*** 添加产品 * @param product*/public synchronized boolean add(T product){if(list.size()<capacity){list.add(product);return true;}return false;}/*** 满* @return*/public synchronized boolean isFull(){if(list.size()>=capacity){return true;}return false;}public synchronized boolean isEmpty(){return list.isEmpty();}public synchronized T get(){if(list.size()>0){return list.remove(0);}return null;}public synchronized int getSize(){return list.size();}public int getCapacity(){return capacity;}
}

Client.java

/*** TODO Comment of Client* @author 百恼 2013-11-7下午12:20:08**/
public class Client {public static void main(String[] args){Object producerMonitor = new Object();Object consumerMonitor = new Object();Container<Bread> container = new Container<Bread>(10);//生产者开动new Thread(new Producer(producerMonitor,consumerMonitor,container)).start();new Thread(new Producer(producerMonitor,consumerMonitor,container)).start();new Thread(new Producer(producerMonitor,consumerMonitor,container)).start();new Thread(new Producer(producerMonitor,consumerMonitor,container)).start();//消费者开动new Thread(new Consumer(producerMonitor,consumerMonitor,container)).start();new Thread(new Consumer(producerMonitor,consumerMonitor,container)).start();}
}

整个生产者消费者模型写到这里就要收尾了，这里可以用concurrent包中的CountDownLatch去实现，原理一样，在控制粒度上会更细一些。有兴趣的同学也可以去实现一下。另外我的模型中有明显的漏洞，代码中也注释出来了，就是有可能出现数据丢失的情况（在把面包加入到容器的过程中）。这又应该怎么处理？