14.多线程设计模式 - Master-Worker模式

多线程设计模式 - Master-Worker模式

 并发设计模式属于设计优化的一部分，它对于一些常用的多线程结构的总结和抽象。与串行相比并行程序结构通常较为复杂，因此合理的使用并行模式在多线程并发中更具有意义。

1. Master-Worker模式

• - Master-Worker模式是常用的并行模式。它的核心思想是系统由两类进程协作工作：Master进程和Worker进程。Master负责接收和分配任务，Worker负责处理子任务。当各个Worker子进程处理完成后，会将结果返回给Master，由Master做归纳和总结。
• - 其好处是能将一个大任务分解成若干个小任务，并行执行，从而提高系统的吞吐量。

示例：下
说明：看类注释即可明了 不明白可以去屎了

//Task.java  
  public class Task {

      private int id;
      private String name;
      private int price;
      public int getId() {
          return id;
      }
      public void setId(int id) {
          this.id = id;
      }
      public String getName() {
          return name;
      }
      public void setName(String name) {
          this.name = name;
      }
      public int getPrice() {
          return price;
      }
      public void setPrice(int price) {
          this.price = price;
      }
  }
  //Worker.java   
  public class Worker implements Runnable{

      private ConcurrentLinkedQueue<Task> workQueue;
      private ConcurrentHashMap<String,Object> resultMap;

      @Override
      public void run() {
          while(true){
              Task input = this.workQueue.poll();
              if(input==null)break;
              //真正的去做业务处理
              Object output = MyWorker.handle(input);
              this.resultMap.put(Integer.toString(input.getId()), output);
          }
      }

      public static Object handle(Task input){
          return null;
      }

      public void setWorkerQueue(ConcurrentLinkedQueue<Task> workQueue) {
          this.workQueue=workQueue;
      }

      public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
          this.resultMap=resultMap;

      }
  }

  //MyWorker.java继承Worker.java，重写handle方法
  public class MyWorker extends Worker{

      public static Object handle(Task input) {
          Object output = null;
          try {
              //处理task的耗时，可能是数据的加工，也可能是操作数据库
              Thread.sleep(500);
              output = input.getPrice();
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
          return output;
      }

  }
  //Master.java *
  public class Master {

      //1 应该有一个承装任务的集合
      private ConcurrentLinkedQueue<Task> workQueue = new  ConcurrentLinkedQueue<Task>();
      //2 使用普通的HashMap去承装所有的worker对象
      private HashMap<String,Thread> workers = new HashMap<String,Thread>();
      //3 使用一个容器承装每一个worker并发执行的结果集
      private ConcurrentHashMap<String,Object> resultMap = new ConcurrentHashMap<String,Object>();
      //4 构造方法
      public Master(Worker worker,int workerCount){
          //每一个worker对象都需要有master的应用workQueue用于任务的领取，resultMap用于任务的提交
          worker.setWorkerQueue(this.workQueue);
          worker.setResultMap(this.resultMap);

          for(int i=0;i<workerCount;i++){
              //key表示每一个worker的名字,value表示线程执行对象
              workers.put("子节点"+Integer.toString(i), new Thread(worker));
          }
      }
      //5 提交方法
      public void submit(Task task){
          this.workQueue.add(task);
      }
      //6 需要有一个执行的方法(启动应用程序让所有的worker工作)
      public void execute(){
          for(Map.Entry<String,Thread> me:workers.entrySet()){
              me.getValue().start();
          }
      }
      //8 判断线程是否执行完毕
      public boolean isComplete() {
          for(Map.Entry<String, Thread> me:workers.entrySet()){
              if(me.getValue().getState()!=Thread.State.TERMINATED){
                  return false;
              }
          }
          return true;
      }
      //9 返回结果集数据
      public int getResult() {
          int ret = 0;
          for(Map.Entry<String,Object> me:resultMap.entrySet()){
              //汇总逻辑
              ret+=(Integer)me.getValue();
          }
          return ret;
      }
  }
  //主函数
  public class Main {

      public static void main(String[] args) {
          System.out.println("我的机器可用processor的数量："+Runtime.getRuntime().availableProcessors());
          Master master = new Master(new MyWorker(),10);

          Random r = new Random();
          for(int i=0;i<100;i++){
              Task task = new Task();
              task.setId(i);
              task.setName("任务"+i);
              task.setPrice(r.nextInt(1000));
              master.submit(task);
          }
          master.execute();

          long start = System.currentTimeMillis();
          while(true){
              if(master.isComplete()){
                  long end = System.currentTimeMillis() -start;
                  int ret = master.getResult();
                  System.out.println("最终结果："+ret+",执行耗时："+end+"毫秒");
                  break;
              }
          }
      }

  }