Java多线程Producer-Consumer模式我来做你来用

概述

生产者安全地将数据交给消费者。当两者在不同的线程运行时，处理速度会引起问题。生产者和消费者一般都有多个，也可以只有一个生产者和一个消费者，称为Pipe模式。

示例程序

3位蛋糕师傅制作蛋糕，房子桌子上，然后3位客人来吃这些蛋糕。
    -  糕点师（MakerThread）制作蛋糕（String），并将其放在桌子上
    -  桌子上最多可放置3个蛋糕
    -  如果桌子上已经放满3个蛋糕时糕点师还要放置蛋糕，必须等到桌子上空出位置
    - 客人（EaterThread）取桌子上的蛋糕吃
    - 客人按蛋糕被放置到桌子上的顺序来取蛋糕
    - 当桌子上1个蛋糕都没有时， 客人若要取蛋糕，必须等到桌子上放置了蛋糕

Main类

 Main类会创建一个桌子的实例，并启动表示蛋糕师和客人的线程。MakerThread和EaterThread的构造函数传入的数字只是用来作为随机数的种子，数值本身并没有什么特别的意义。

public class Main {
  public static void main(String[] args){
    Table table = new Table(3);     //创建能放三个蛋糕的桌子
    new MakerThread("MakerThread-1", table, 12121).start();
    new MakerThread("MakerThread-2", table, 24242).start();
    new MakerThread("MakerThread-3", table, 36363).start();
    new EaterThread("EaterThread-1", table, 47474).start();
    new EaterThread("EaterThread-2", table, 58585).start();
    new EaterThread("EaterThread-3", table, 69696).start();
  }
}

MakerThread类

import java.util.Random;
public class MakerThread extends Thread {
  private final Random random;
  private final Table table;
  private static int id = 0;          //蛋糕的流水号（所有蛋糕师公用）
  public MakerThread(String name, Table table, long seed){
    super(name);
    this.table = table;
    this.random = new Random(seed);
  }
  public void run() {
    try {
      while (true){
        Thread.sleep(random.nextInt(1000));
        String cake = "[ Cake No." + nextId() + " by " + getName() + "  ]";
         table.put(cake);
      }
    } catch (InterruptedException e){
    }
  }
  private static synchronized int nextId(){
    return id++;
  }
}

EaterThread 类

import java.util.Random;
public class EaterThread extends Thread {
  private final Random random;
  private final Table table;
  public EaterThread(String name, Table table, long seed){
    super(name);
    this.table = table;
    this.random = new Random(seed);
  }
  public void run(){
    try{
      while(true){
        String cake = table.take();
        Thread.sleep(random.nextInt(1000));
      }
    } catch (InterruptedException e){}
  }
}

Table 类

public class Table {
  private final String[] buffer;
  private int tail;
  private int head;
  private int count;
  public Table(int count){
    this.buffer = new String[count];
    this.head = 0;
    this.tail = 0;
    this.count = 0;
  }
  //放置蛋糕
  public synchronized void put(String cake) throws InterruptedException {
    System.out.println(Thread.currentThread().getName() +"  puts " + cake);
    while(count >= buffer.length){
      wait();
    }
    buffer[tail] = cake;
    tail = (tail + 1) % buffer.length;
    count++;
    notifyAll();
  }
  //拿取蛋糕
  public synchronized String take() throws InterruptedException {
    while (count <= 0) {
      wait();
    }
    String cake = buffer[head];
    head = ( head + 1) % buffer.length;
    count--;
    notifyAll();
    System.out.println(Thread.currentThread().getName() + " takes " + cake;
    return cake
  }
}

put方法解读

• throws InterruptedException
  put方法被声明为可能抛出异常InterruptedException异常的方法。当看到throwsInterruptedException时， 我们可以理解为“该方法可以取消”。
• Guarded Suspension模式
  put方法，使用while条件表达的守护条件，即“当前桌子上放置的蛋糕数小于能够放置的最大数”，也就是“还有可以放置的位置”，作为put方法的守护条件。
• tail 和 count 的更新
  tail表示下次放置蛋糕的位置， buffer[tail] = cake; 放置后，tail要加1就可以了 tail = (tail + 1) % buffer.length;
  桌子上的蛋糕增多了，所以count的值也需要加1。
• notifyAll
  上面已经把蛋糕放到桌子上，由于桌子的状态发生了变化，所以要执行notifyAll，唤醒所有正在wait的线程。

take方法解读

• throws InterruptedException
  说明此方法时可以取消的方法
• Guarded Suspension 模式
  守护条件是“当前桌子上至少有一个蛋糕”。
• head 和 count的更新
  head 表示下次取蛋糕的位置，蛋糕取走之后，head也要前进。head = (head + 1) % buffer.length;
  桌子上的蛋糕取走了一个， 所以count的值也需要减一。
• notifyAll
  通过上面的处理，蛋糕被取走了，桌面就发生了变化，所以要执行notifyAll，唤醒所有正在wait的线程。

Producer-Consumer 模式的角色

• Data 由Producer角色生成，供Consumer角色使用。在示例程序中，由String类（蛋糕）扮演。
• Producer 此角色生成Data角色，并将其传递给Channel角色。在示例中，由MakerThread扮演。
• Consumer 消费者角色从Channel角色获取Data角色。在示例程序中，由EaterThread扮演此角色
• Channel 通道角色用于承担传递Data角色的中转站、通道的任务。在示例程序中由Table扮演。