Question

生产者-消费者模式是一个经典的多线程设计模式，它为多线程间的协作提供了良好的解决方案。在生产者-消费者模式中，通常有两类线程，即若干个生产者线程和若干个消费者线程。生产者线程负责提交用户请求，消费者线程则负责具体处理生产者提交的任务。生产者和消费者之间则通过共享内存缓冲区进行通信。

如图5.1所示，展示了生产者-消费者模式的基本结构。三个生产者线程将任务提交到共享内存缓冲区，消费者线程并不直接与生产者线程通信，而在共享内存缓冲区中获取任务，并进行处理。

图5.1　生产者-消费者模式架构图

注意：生产者-消费者模式中的内存缓存区的主要功能是数据在多线程间的共享，此外，通过该缓冲区，可以缓解生产者和消费者间的性能差。

生产者-消费者模式的核心组件是共享内存缓存区，它作为生产者和消费者间的通信桥梁，避免了生产者和消费者的直接通信，从而将生产者和消费者进行解耦。生产者不需要知道消费者的存在，消费者也不需要知道生产者的存在。

同时，由于内存缓冲区的存在，允许生产者和消费者在执行速度上存在时间差，无论是生产者在某一局部时间内速度高于消费者，还是消费者在局部时间内高于生产者，都可以通过共享内存缓冲区得到缓解，确保系统正常运行。

生产者-消费者模式的主要角色如表5.1所示。

表5.1　生产者-消费者模式主要角色

图5.2显示了生产者-消费者模式一种实现的具体结构。

图5.2　生产者-消费者实现类图

其中，BlockigQueue充当了共享内存缓冲区，用于维护任务或数据队列（PCData对象）。我强烈建议大家先回顾一下第3章有关BlockingQueue的相关知识，对于理解整个生产者和消费者结构有重要的帮助。PCData对象表示一个生产任务，或者相关任务的数据。生产者对象和消费者对象均引用同一个BlockigQueue实例。生产者负责创建PCData对象，并将它加入BlockigQueue中，消费者则从BlockigQueue队列中获取PCData。

基于图5.2所示结构，实现一个基于生产者-消费者模式的求整数平方的并行程序。

首先，生产者线程的实现如下，它构建PCData对象，并放入BlockingQueue队列中。

public class Producer implements Runnable {
  private volatile boolean isRunning = true;
  private BlockingQueue＜PCData＞ queue;             //内存缓冲区
  private static AtomicInteger count = new AtomicInteger();  //总数，原子操作
  private static final int SLEEPTIME = 1000;

  public Producer(BlockingQueue＜PCData＞ queue) {
    this.queue = queue;
  }

  public void run() {
    PCData data = null;
    Random r = new Random();

    System.out.println("start producer id="+Thread.currentThread().getId());
    try {
      while (isRunning) {
        Thread.sleep(r.nextInt(SLEEPTIME));
        data = new PCData(count.incrementAndGet());    //构造任务数据
        System.out.println(data+" is put into queue");
        if (!queue.offer(data, 2, TimeUnit.SECONDS)) {   //提交数据到缓冲区中
          System.err.println("failed to put data：" + data);
        }
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
      Thread.currentThread().interrupt();
    }
  }
  public void stop() {
    isRunning = false;
  }
}

对应的消费者的实现如下。它从BlockingQueue队列中取出PCData对象，并进行相应的计算。

public class Consumer implements Runnable {
  private BlockingQueue＜PCData＞ queue;           //缓冲区
  private static final int SLEEPTIME = 1000;

  public Consumer(BlockingQueue＜PCData＞ queue) {
    this.queue = queue;
  }

  public void run() {
    System.out.println("start Consumer id="
        + Thread.currentThread().getId());
    Random r = new Random();              //随机等待时间

    try {
      while(true){
        PCData data = queue.take();         //提取任务
        if (null != data) {
          int re = data.getData() * data.getData();  //计算平方
          System.out.println(MessageFormat.format("{0}*{1}={2}",
              data.getData(), data.getData(), re));
          Thread.sleep(r.nextInt(SLEEPTIME));
        }
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
      Thread.currentThread().interrupt();
    }
  }
}

PCData作为生产者和消费者之间的共享数据模型，定义如下：

public final class PCData {                //任务相关的数据
  private  final int intData;              //数据
  public PCData(int d){
    intData=d;
  }
  public PCData(String d){
    intData=Integer.valueOf(d);
  }
  public int getData(){
    return intData;
  }
  @Override
  public String toString(){
    return "data:"+intData;
  }
}

在主函数中，创建三个生产者和三个消费者，并让它们协作运行。在主函数的实现中，定义LinkedBlockingQueue作为BlockingQueue的实现类。

public class Main {
  public static void main(String[] args) throws InterruptedException {
    //建立缓冲区
    BlockingQueue＜PCData＞ queue = new LinkedBlockingQueue＜PCData＞(10);
    Producer producer1 = new Producer(queue);          //建立生产者
    Producer producer2 = new Producer(queue);
    Producer producer3 = new Producer(queue);
    Consumer consumer1 = new Consumer(queue);          //建立消费者
    Consumer consumer2 = new Consumer(queue);
    Consumer consumer3 = new Consumer(queue);
    ExecutorService service = Executors.newCachedThreadPool();   //建立线程池
    service.execute(producer1);                //运行生产者
    service.execute(producer2);
    service.execute(producer3);
    service.execute(consumer1);                //运行消费者
    service.execute(consumer2);
    service.execute(consumer3);
    Thread.sleep(10 * 1000);
    producer1.stop();                      //停止生产者
    producer2.stop();
    producer3.stop();
    Thread.sleep(3000);
    service.shutdown();
  }
}

注意：生产者-消费者模式很好地对生产者线程和消费者线程进行解耦，优化了系统整体结构。同时，由于缓冲区的作用，允许生产者线程和消费者线程存在执行上的性能差异，从一定程度上缓解了性能瓶颈对系统性能的影响。