Java 生产者模式 消费者模式

1 // The standard idiom for calling the wait
2 synchronized(sharedObject) {
3        while(condition){
4             sharedObject.wait();// Releases lock, and reacquires on wake up
5        }
6        // do action based upon condition e.g. take or put into queue
7 }

使用wait和notify函数的规范代码模版。

在while循环里使用wait的目的：是在线程被唤醒前后都持续检查条件是否被满足，如果条件并未改变，wait被调用之前notify的唤醒通知就来了，那个这个线程并不能保证被唤醒，有坑会导致死锁的问题。

 1 public class ProducerConsumer {
 2       public static void main (String args[]) {
 3             Queue<Integer> buffer = new LinkedList<>();
 4             int maxSize = 10;
 5             Thread producer = new Producer(buffer, maxSize, "PRODUCER");
 6             Thread consumer = new Producer(buffer, maxSize, "CONSUMER");
 7             producer.start();
 8             consumer.start();
 9       }
10 }    

在上面代码中，声明了一个LinkedList作为缓冲区队列（在java中，LinkedList实现了队列的接口）。

class Producer extends Thread  {
      private Queue<Integer> queue;
      private int maxSize;
      public Producer(Queue<Integer> queue, int maxSize, String name) {
             super(name);
             this.queue = queue;
             this.maxSize = maxSize;
      }

      @Override
      public void run () {
           while (true) {
                 synchronized (queue) {  // 其它线程不能在我们检查条件时改变这个队列
                        while (queue.size() == maxSize) {
                              try {
                                    // Queue is full
                                    // Producer thread waiting for
                                    // consumer to take something from queue
                                    queue.wait();
                              } catch (Exception e) {
                                    e.printStackTrace();
                              }
                              Random random = new Random();
                              int producerValue = random.nextInt();
                              queue.add(producerValue);
                              queue.notifyAll();
                        }
                 }
           }
      }
}

如上代码为生产者，其在无限循环中持续往LinkedList里插入随机整数直到LinkedList满。如果队列满了，那么生产者线程会在消费者线程消耗掉队列里的任意一个整数，并用notify通知生产者线程之前持续等待。

 1 class Consumer extends Thread {
 2     private Queue<Integer> queue;
 3     private int maxSize;
 4     public Consumer (Queue<Integer> queue, int maxSize, String name ){
 5          super(name);
 6          this.queue = queue;
 7          this.maxSize= maxSize;
 8     }
 9     @Override
10     public void run() {
11           while (true) {
12                 synchronized (queue) {
13                       while (queue.isEmpty()) {
14                              // queue is empty;
15                              // Consumer thread is waiting
16                              // for producer thread to put something in queue
17                              try {
18                                    queue.wait();
19                              } catch (Exception e) {
20                                    e.printStackTrace();
21                              }
22                              print("Consumer value : " + queue remove());
23                              queue.notifyAll();
24                       }
25                 }
26           }
27     }
28 }

注意：

1. 你可以使用wait和notify函数来实现线程间通信，你可以用它们来实现多线程之间的通信。

2. 永远在synchronized的函数或者对象里使用wait、notify、notifyAll，不然java虚拟机会生成IllgalMonitorStateException。

3. 永远在while循环里而不是在if语句下使用wait，这样，循环会在线程睡眠前后都检查wait的条件，并在条件实际上并未改变的情况下处理唤醒通知。

4. 永远在多线程间共享的对象（生产者消费者模型里即缓冲区队列）上使用wait。

Producer - Consumer Pattern:

Consumer:

sunchronized (obj) {
     while (!workToDo) {
           obj.wait();
     }
     // get next item from this queue
     workToDo = false;
}
// do work on the item

Producer:

1 synchronized (obj) {
2       if (!workToDo) {
3            // add work to queue
4            workToDo = true;
5       }
6       obj.notifyAll();
7 }