JAVA多线程（三）生产者消费者模式及实现方法

系统管理员 2022-05-24 03:06 127阅读 0赞

本文介绍两种实现生产者消费者模式的方法。

# 1 生产者消费者模式 #

首先来了解什么是生产者消费者模式。该模式也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

要解决该问题，就必须让生产者在缓冲区满时休眠（要么干脆就放弃数据），等到下次消费者消耗缓冲区中的数据的时候，生产者才能被唤醒，开始往缓冲区添加数据。同样，也可以让消费者在缓冲区空时进入休眠，等到生产者往缓冲区添加数据之后，再唤醒消费者。通常采用进程间通信的方法解决该问题，常用的方法有信号灯法等。如果解决方法不够完善，则容易出现死锁的情况。出现死锁时，两个线程都会陷入休眠，等待对方唤醒自己。该问题也能被推广到多个生产者和消费者的情形。

# 2 用wait和notify实现 #

这种方法的基本原理是：一个线程负责生产数据，放到共享区域，然后通知另一个线程去消耗数据。如果没有wait()和notify(),消费者线程就要不停去检查是否有数据被产生。

接下来介绍一下wait()和notify()，在这里把它们和sleep()做一个对比，方便理解

<table style="border-spacing:0px;width:852px;text-align:center;color:rgb(51,51,51);font-family:'-apple-system', 'SF UI Text', Arial, 'PingFang SC', 'Hiragino Sans GB', 'Microsoft YaHei', 'WenQuanYi Micro Hei', sans-serif, SimHei, SimSun;font-size:14px;background-color:rgb(255,255,255);"> 
 <thead> 
  <tr> 
   <th style="margin:0px;vertical-align:middle;">不同点</th> 
   <th align="left" style="margin:0px;vertical-align:middle;">sleep()</th> 
   <th style="margin:0px;vertical-align:middle;">wait()和notify()</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="margin:0px;vertical-align:middle;">原理</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">线程用来控制自身流程，会使该线程暂停执行一段时间，把执行机会让给其它线程。时间一到就复苏。</td> 
   <td style="margin:0px;vertical-align:middle;">是Object类的方法，会使当前拥有该对象锁的进程等待，直到其他线程调用notify()方法。</td> 
  </tr> 
  <tr> 
   <td style="margin:0px;vertical-align:middle;">锁的处理机制</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">只是让线程暂停执行一段时间，不会释放锁</td> 
   <td style="margin:0px;vertical-align:middle;">调用wait()，线程会释放掉锁</td> 
  </tr> 
  <tr> 
   <td style="margin:0px;vertical-align:middle;">使用区域</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">必须放在同步控制方法或者同步语句块中</td> 
   <td style="margin:0px;vertical-align:middle;">可以放在任何地方</td> 
  </tr> 
  <tr> 
   <td style="margin:0px;vertical-align:middle;">异常</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">必须捕获异常，例如InterruptedException等</td> 
   <td style="margin:0px;vertical-align:middle;">不用捕获异常</td> 
  </tr> 
 </tbody> 
</table>

sleep不会释放锁，容易导致死锁（在我的上一篇博客[ JAVA多线程（二）竞态条件、死锁及同步机制][JAVA]有描述）。因此推荐使用wait()和notify()。下面转载一份[源代码][Link 1],

import java.util.LinkedList;
    import java.util.Queue;
    import org.apache.log4j.Logger;
    
    public class InterThreadCommunicationExample { 
    
        public static void main(String args[]) {
    
            final Queue sharedQ = new LinkedList();
    
            Thread producer = new Producer(sharedQ);
            Thread consumer = new Consumer(sharedQ);
    
            producer.start();
            consumer.start();
    
        }
    }
    
    public class Producer extends Thread { 
        private static final Logger logger = Logger.getLogger(Producer.class);
        private final Queue sharedQ;
    
        public Producer(Queue sharedQ) {
            super("Producer");
            this.sharedQ = sharedQ;
        }
    
        @Override
        public void run() {
    
            for (int i = 0; i < 4; i++) {
    
                synchronized (sharedQ) {
                    //waiting condition - wait until Queue is not empty
                    while (sharedQ.size() >= 1) {
                        try {
                            logger.debug("Queue is full, waiting");
                            sharedQ.wait();
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                    }
                    logger.debug("producing : " + i);
                    sharedQ.add(i);
                    sharedQ.notify();
                }
            }
        }
    }
    
    public class Consumer extends Thread { 
        private static final Logger logger = Logger.getLogger(Consumer.class);
        private final Queue sharedQ;
    
        public Consumer(Queue sharedQ) {
            super("Consumer");
            this.sharedQ = sharedQ;
        }
    
        @Override
        public void run() {
            while(true) {
    
                synchronized (sharedQ) {
                    //waiting condition - wait until Queue is not empty
                    while (sharedQ.size() == 0) {
                        try {
                            logger.debug("Queue is empty, waiting");
                            sharedQ.wait();
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                    }
                    int number = sharedQ.poll();
                    logger.debug("consuming : " + number );
                    sharedQ.notify();
    
                    //termination condition
                    if(number == 3){
       break; }
                }
            }
        }
    }
    
    Output:
    05:41:57,244 0    [Producer] DEBUG concurrency.Producer  - producing : 0
    05:41:57,260 16   [Producer] DEBUG concurrency.Producer  - Queue is full, waiting
    05:41:57,260 16   [Consumer] DEBUG concurrency.Consumer  - consuming : 0
    05:41:57,260 16   [Consumer] DEBUG concurrency.Consumer  - Queue is empty, waiting
    05:41:57,260 16   [Producer] DEBUG concurrency.Producer  - producing : 1
    05:41:57,260 16   [Producer] DEBUG concurrency.Producer  - Queue is full, waiting
    05:41:57,260 16   [Consumer] DEBUG concurrency.Consumer  - consuming : 1
    05:41:57,260 16   [Consumer] DEBUG concurrency.Consumer  - Queue is empty, waiting
    05:41:57,260 16   [Producer] DEBUG concurrency.Producer  - producing : 2
    05:41:57,260 16   [Producer] DEBUG concurrency.Producer  - Queue is full, waiting
    05:41:57,260 16   [Consumer] DEBUG concurrency.Consumer  - consuming : 2
    05:41:57,260 16   [Consumer] DEBUG concurrency.Consumer  - Queue is empty, waiting
    05:41:57,260 16   [Producer] DEBUG concurrency.Producer  - producing : 3
    05:41:57,276 32   [Consumer] DEBUG concurrency.Consumer  - consuming : 3

# 3 使用阻塞队列实现 #

使用wait()和notify()是经典方法，这里介绍一个高级方法。

BlockingQueue中提供了put()和take()方法，可以极大简化生产者消费者模式的实现过程。这一过程的基本原理是，如果队列满了，put()方法就会被阻塞；如果队列是空的，take()方法会阻塞。与传统的wait()和notify()方法相比，使用阻塞队列更简单，更便于理解。下面是一个简单的例子：

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.logging.Level;
    import java.util.logging.Logger;
    
    public class ProducerConsumerPattern { 
    
        public static void main(String args[]){
    
         //Creating shared object
         BlockingQueue sharedQueue = new LinkedBlockingQueue();
    
         //Creating Producer and Consumer Thread
         Thread prodThread = new Thread(new Producer(sharedQueue));
         Thread consThread = new Thread(new Consumer(sharedQueue));
    
         //Starting producer and Consumer thread
         prodThread.start();
         consThread.start();
        }
    
    }
    
    //Producer Class in java
    class Producer implements Runnable { 
    
        private final BlockingQueue sharedQueue;
    
        public Producer(BlockingQueue sharedQueue) {
            this.sharedQueue = sharedQueue;
        }
    
        @Override
        public void run() {
            for(int i=0; i<10; i++){
                try {
                    System.out.println("Produced: " + i);
                    sharedQueue.put(i);
                } catch (InterruptedException ex) {
                    Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    
    }
    
    //Consumer Class in Java
    class Consumer implements Runnable{ 
    
        private final BlockingQueue sharedQueue;
    
        public Consumer (BlockingQueue sharedQueue) {
            this.sharedQueue = sharedQueue;
        }
    
        @Override
        public void run() {
            while(true){
                try {
                    System.out.println("Consumed: "+ sharedQueue.take());
                } catch (InterruptedException ex) {
                    Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    
    
    }
    
    Output:
    Produced: 0
    Produced: 1
    Consumed: 0
    Produced: 2
    Consumed: 1
    Produced: 3
    Consumed: 2
    Produced: 4
    Consumed: 3
    Produced: 5
    Consumed: 4
    Produced: 6
    Consumed: 5
    Produced: 7
    Consumed: 6
    Produced: 8
    Consumed: 7
    Produced: 9
    Consumed: 8
    Consumed: 9

在后面的一篇博客中，给本文提供了补充的例子。[用lock、synchronized、阻塞队列三种方法实现生产者消费者模式][lock_synchronized]，实现的内容是生产者产生随机数（为了方便阅读结果，我把随机数限定在10以内的整数），消费者读取并打印。

[JAVA]: http://blog.csdn.net/antony9118/article/details/51480978
[Link 1]: http://javarevisited.blogspot.sg/2013/12/inter-thread-communication-in-java-wait-notify-example.html
[lock_synchronized]: http://blog.csdn.net/antony9118/article/details/51500278