第三部分-并发设计模式36:生产者-消费者模式

1.生产者-消费者模式

Worker Thread模式类比现实工厂里车间工人的工作模式
生产者-消费者模式，类比显示工厂里流水线模式

2.应用场景

Log4j2异步Appender内部使用生产者-消费者模式。

3.生产者，消费者模式优点

核心：任务队列
生产者线程生产任务，并将任务添加到任务队列中
消费者线程从任务队列中获取任务并执行

（1）架构设计方面，优点是解耦。解耦的关键是组件之间的依赖关系和通信方式必须受限。
生产者-消费者模式中，生产者和消费者没有任何依赖关系，之间通信只能通过任务队列。
（2）支持异步，并且能够平衡生产者和消费者的速度差异。生产者线程只需要将任务添加到任务队列而无需等待任务被消费者执行完。任务的生产和消费是异步的。传统的方法调用时同步的。
（3）任务队列究竟是什么用，平衡生产者和消费者的速度差异。加入生产速率慢，3个生产者，1个消费者。比例1:3。如果采用创建新线程方式，会创建3个子线程。生产者，消费者模式，消费线程只需要1个就可以了。java中的线程是重的，映射到系统的线程的。线程太多，增加上下文切换成本。所以线程不是越多越好。

4.批量执行提升性能

如果使用 轻量级线程，没有必要平衡生产者和消费者的速度差异了，
轻量级线程，本身就是廉价的。
批量执行任务上，使用生产者-消费者模式灰常好使

5.批量执行场景

类比1000条sql的insert语句，方案1，采用1000个线程并发执行，每个线程insert 1条数据，方案2是，采用1个线程，拉取下来1000条语句，合并成1条sql语句，一次性插入到数据库中去。

伪代码

//任务队列
BlockingQueue<Task> bq=new
  LinkedBlockingQueue<>(2000);
//启动5个消费者线程
//执行批量任务  
void start() {
  ExecutorService es=executors
    .newFixedThreadPool(5);
  for (int i=0; i<5; i++) {
    es.execute(()->{
      try {
        while (true) {
          //获取批量任务
          List<Task> ts=pollTasks();
          //执行批量任务
          execTasks(ts);
        }
      } catch (Exception e) {
        e.printStackTrace();
      }
    });
  }
}
//从任务队列中获取批量任务
List<Task> pollTasks() 
    throws InterruptedException{
  List<Task> ts=new LinkedList<>();
  //阻塞式获取一条任务
  Task t = bq.take();
  while (t != null) {
    ts.add(t);
    //非阻塞式获取一条任务
    t = bq.poll();
  }
  return ts;
}
//批量执行任务
execTasks(List<Task> ts) {
  //省略具体代码无数
}

说明:创建了5个消费者线程负责批量执行sql，
pollTasks()中，首先以阻塞方式获取任务队列中的一条任务，而后以非阻塞方式获取任务。采取阻塞方式，是因为如果任务队列中没有任务，这样能避免无限循环。

6.分段提交提升性能

分阶段提交的应用场景。
写文件刷盘会慢，不是很重要的数据，异步刷盘方式。

例如日志组件:
ERROR级别日志立即刷盘
数据积累到500条立即刷盘
存在未刷盘数据，且5秒内未曾刷盘，立即刷盘

伪代码

class Logger {
  //任务队列  
  final BlockingQueue<LogMsg> bq
    = new BlockingQueue<>();
  //flush批量  
  static final int batchSize=500;
  //只需要一个线程写日志
  ExecutorService es = 
    Executors.newFixedThreadPool(1);
  //启动写日志线程
  void start(){
    File file=File.createTempFile(
      "foo", ".log");
    final FileWriter writer=
      new FileWriter(file);
    this.es.execute(()->{
      try {
        //未刷盘日志数量
        int curIdx = 0;
        long preFT=System.currentTimeMillis();
        while (true) {
          LogMsg log = bq.poll(
            5, TimeUnit.SECONDS);
          //写日志
          if (log != null) {
            writer.write(log.toString());
            ++curIdx;
          }
          //如果不存在未刷盘数据，则无需刷盘
          if (curIdx <= 0) {
            continue;
          }
          //根据规则刷盘
          if (log!=null && log.level==LEVEL.ERROR ||
              curIdx == batchSize ||
              System.currentTimeMillis()-preFT>5000){
            writer.flush();
            curIdx = 0;
            preFT=System.currentTimeMillis();
          }
        }
      }catch(Exception e){
        e.printStackTrace();
      } finally {
        try {
          writer.flush();
          writer.close();
        }catch(IOException e){
          e.printStackTrace();
        }
      }
    });  
  }
  //写INFO级别日志
  void info(String msg) {
    bq.put(new LogMsg(
      LEVEL.INFO, msg));
  }
  //写ERROR级别日志
  void error(String msg) {
    bq.put(new LogMsg(
      LEVEL.ERROR, msg));
  }
}
//日志级别
enum LEVEL {
  INFO, ERROR
}
class LogMsg {
  LEVEL level;
  String msg;
  //省略构造函数实现
  LogMsg(LEVEL lvl, String msg){}
  //省略toString()实现
  String toString(){}
}

说明:调用info()和error()方法写日志，创建LogMsg对象，并加入到阻塞队列中
调用info()和error()方法的线程是生产者。而真正日志写入文件的是消费者线程。
在消费者线程中，根据刷盘规则执行刷盘操作。

7.总结

分布式场景下，借助MQ来实现生产者-消费者模式。
MQ支持两种消息模式，一种点对点，一个是发布订阅。点对点里一个消息只会被一个消费者消费
发布订阅模式，一个消息会被多个消费者消费，消息广播

8.生产案例

生产者消费者模式
Worker 类，封装生产，队列，消费3个角色

public abstract class Worker implements Comparable<Worker> {

    private static final Logger logger = LoggerFactory.getLogger(Worker.class);

    private Storage<Record> storage; //队列
    private Sinker<Record> sinker; // 消费者
    private FetcherThread fetcherThread; // 生产者

    private volatile boolean stop = false;

    public void init() throws Exception {
        sinker = new CommonSinker(getProcessor(), getCommit(),this); // 创建消费者
        Properties prop = new Properties();
        prop.setProperty("platform", getPlatform().name());
        prop.setProperty("resType",getResourceType());
        sinker.init(prop); 
        fetcherThread = new FetcherThread(this); // 创建生产者线程
        storage = new RAMStorage<>(); // 创建队列
    }

    abstract String getResourceType();

    public Platform getPlatform() {
        return Platform.UNKNOWN;
    }

    abstract Fetcher<Record> getFetcher();

    abstract List<Processor> getProcessor();

    abstract CommitProcessor getCommit();

    public boolean hasStoped(){
        return this.stop;
    }

    // 应用并发设计模式35: 两阶段终止模式（停止位及interrupt）
    public void stop() {
        try {
            this.stop = true;
            fetcherThread.stop();
            sinker.stop();
            storage.setPushClosed(true);
            storage = null;
        }catch (Throwable ignore){}
    }

    public void run() {
        new Thread(fetcherThread).start();  // 生产者开始生产
        try {
            sinker.sinkDataFromStorage(storage); // 消费者从storage中消费消息
        } catch (Exception ex) {
            WorkAlarm.workAlarm("sinker failed "+ ex.getMessage());
            logger.error("sinker failed:", ex);
            this.stop();
        }
    }

    @Override
    public int compareTo(Worker o) {
        if (o == null) {
            return -1;
        }
        return this.getClass().getName().compareTo(o.getClass().getName());
    }

    public class FetcherThread implements Runnable {
        private Thread current = null;
        private Worker worker;
        public FetcherThread(Worker worker) {
            this.worker = worker;
        }

        @Override
        public void run() {
            try {
                current = Thread.currentThread();
                getFetcher().fetchToCache(storage); // 生产者往队列中塞消息
                storage.setPushClosed(true);
            } catch (Exception e) {
                WorkAlarm.workAlarm("fetcher failed:["+getFetcher().getClass().getSimpleName()+"],"+e.getMessage());
                worker.stop();
                logger.error("fetcher failed :", e);
            }
        }

        // 应用并发设计模式35: 两阶段终止模式（停止位及interrupt）
        public void stop() {
            getFetcher().stop();
            current.interrupt();
        }
    }
}

生产者

public class XxxWorker extends Worker {

    private final Fetcher<Record> fetcher;
    private final List<Processor> processors;
    private final CommitProcessor commit;

    public JoneWorker() {
        this.fetcher = new XxxFetcher();
        this.commit = new SysCommitProcessor();
        this.processors = Lists.newArrayList(new XxxProcessor());
    }

    @Override
    String getResourceType() {
        return "";
    }

    @Override
    public Platform getPlatform() {
        return Platform.Xxx;
    }

    @Override
    Fetcher<Record> getFetcher() {
        return fetcher;
    }

    @Override
    List<Processor> getProcessor() {
        return processors;
    }

    @Override
    CommitProcessor getCommit() {
        return commit;
    }
}

消费者（仅仅消费消息，然后交给处理器processor进行处理）

public interface Sinker<T> {
    void sinkDataFromStorage(Storage<T> storage) throws Exception;
    void init(Properties prop) throws Exception;
    void stop();
}

public class CommonSinker implements Sinker<Record> {

    private final List<Processor> processors;

    private final CommitProcessor commitProcessor;

    private Worker worker;

    public CommonSinker(List<Processor> processors, CommitProcessor commitProcessor,Worker worker) {
        this.processors = processors;
        this.commitProcessor = commitProcessor;
        this.worker = worker;
    }

    @Override
    public void sinkDataFromStorage(Storage<Record> storage) throws Exception {
        Record l;
        while ((l = storage.pull()) != null && !worker.hasStoped()) {
            Record[] result = new Record[]{l};
            for (Processor processor : processors) {
                if (worker.hasStoped()){
                    return;
                }
                result = processor.process(result);
                if (result == null) {
                    break;
                }
            }
            if (!worker.hasStoped()){
                commitProcessor.process(l);
            }
            if (commitProcessor.getSize() >= 50 && !worker.hasStoped()) {
                commitProcessor.commit();
            }
        }
        if (!worker.hasStoped()){
            if (commitProcessor.getSize() != 0) {
                commitProcessor.commit();
            }
        }
        if (!worker.hasStoped()){
            commitProcessor.finished();
        }
    }

    @Override
    public void init(Properties prop) throws Exception {
        commitProcessor.init(prop);
    }

    @Override
    public void stop() {
        try {
            for (Processor processor : processors) {
                processor.release();
            }
            commitProcessor.release();
        } catch (Throwable ignore) {
        }
    }
}

封装java队列父类Storage

public abstract class Storage<T> {

    public abstract boolean push(T data) throws Exception;

    public abstract T pull();

    private boolean pushClosed;

    private int limit = 0;

    public void init( int limit) {
        pushClosed = false;
        this.limit = limit;
    }

    public boolean isPushClosed() {
        return pushClosed;
    }

    public void setPushClosed(boolean close) {
        pushClosed = close;
    }

    public int getLimit(){
        return this.limit;
    }
}

public class RAMStorage<T> extends Storage<T> {

    private BlockingQueue<T>  cache;

    public RAMStorage() {
        init(2<<11);
        this.cache = new ArrayBlockingQueue<>(2<<11); // 内嵌有界阻塞队列
    }

    @Override
    public boolean push(T data) throws Exception {
        if (isPushClosed())
            return false;
        cache.put(data);
        return true;
    }

    @Override
    public T pull() {
        T line = null;
        while(true) {
            try {
                line = cache.poll(1, TimeUnit.SECONDS);
                if (line == null && isPushClosed())
                    return cache.poll();
                else if (line != null)
                    return line;
            } catch (InterruptedException e) {
                return null;
            }
        }
    }
}