并行设计模式-- Master-Worker模式

Posted 2020-09-01 星火燎原智勇

　　Java多线程编程中，常用的多线程设计模式包括：Future模式、Master-Worker模式、Guarded Suspeionsion模式、不变模式和生产者-消费者模式等。这篇文章主要讲述Master-Worker模式，关于其他多线程设计模式的地址如下：
　　关于Future模式的详解： 并行设计模式（一）-- Future模式
　　关于Guarded Suspeionsion模式的详解： 并行设计模式（三）-- Guarded Suspeionsion模式
　　关于不变模式的详解： 并行设计模式（四）-- 不变模式
　　关于生产者-消费者模式的详解：并行设计模式（五）-- 生产者-消费者模式

 1. Master-Worker模式

　　Master-Worker模式是常用的并行模式之一，它的核心思想是：系统由两类进程协同工作，即Master进程和Worker进程，Master负责接收和分配任务，Wroker负责处理子任务。当各个Worker进程将子任务处理完成后，将结果返回给Master进程，由Master进程进行汇总，从而得到最终的结果，其具体处理过程如下图所示。

 

　　Master-Worker 模式的好处，它能够将一个大任务分解成若干个小任务并行执行，从而提高系统的吞吐量。而对于系统请求者 Client 来说，任务一旦提交，Master进程会分配任务并立即返回，并不会等待系统全部处理完成后再返回，其处理过程是异步的。因此，Client 不会出现等待现象。

2. Master-Worker模式结构

　　Master-Worker 模式的结构相对比较简单，Master 进程为主要进程，它维护了一个Worker 进程队列、子任务队列和子结果集、Worker 进程队列中的 Worker 进程，不停地从任务队列中提取要处理的子任务，并将子任务的处理结果写入结果集。具体的结构图如下所示：

注意：Master-Worker 模式是一种使用多线程进行数据处理的结构。多个 Worker 进程协作处理用户请求，Master 进程负责维护 Worker 进程，并整合最终处理结果。

3. 代码实现

　　Master-Worker 主要角色分配如下所示：

角 色	作 用
Worker	用于实际处理一个任务
Master	用于任务的分配和最终结果的合成
Main	启动系统，调用开启Master

下面是一个简易版的 Master-Worker 框架 Java 代码实现

1. Master 部分源码实现：

 1 import java.util.HashMap;
 2 import java.util.Map;
 3 import java.util.Queue;
 4 import java.util.concurrent.ConcurrentHashMap;
 5 import java.util.concurrent.ConcurrentLinkedQueue;
 6 
 7 public class Master {
 8     // 任务队列
 9     protected Queue<Object> workQueue = new ConcurrentLinkedQueue<Object>();
10     // Worker进程队列
11     protected Map<String, Thread> threadMap = new HashMap<String, Thread>();
12     // 子任务处理结果集
13     protected Map<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
14     // 构造函数
15     public Master(Worker worker, int countWorker) {
16         worker.setWorkQueue(workQueue); //添加任务队列
17         worker.setResultMap(resultMap); //添加计算结果集合
18         for(int i=0; i<countWorker; i++) {
19             threadMap.put(Integer.toString(i), new Thread(worker, Integer.toString(i))); //循环添加任务进程
20         }
21     }
22     
23     //是否所有的子任务都结束了
24     public boolean isComplete() {
25         for(Map.Entry<String, Thread> entry : threadMap.entrySet()) {
26             if(entry.getValue().getState() != Thread.State.TERMINATED)
27                 return false; //存在未完成的任务
28         }
29         return true;
30     }
31     
32     //提交一个子任务
33     public void submit(Object job) {
34         workQueue.add(job);
35     }
36     
37     //返回子任务结果集
38     public Map<String, Object> getResultMap() {
39         return resultMap;
40     }
41     
42     //执行所有Worker进程，进行处理
43     public void execute() {
44         for(Map.Entry<String, Thread> entry : threadMap.entrySet()) {
45             entry.getValue().start();
46         }
47     }
48 }

2. Worker 进程的源代码实现

 1 import java.util.Map;
 2 import java.util.Queue;
 3 
 4 public class Worker  implements Runnable{
 5     //任务队列，用于取得子任务
 6     protected Queue<Object> workQueue;
 7     //子任务处理结果集
 8     protected Map<String ,Object> resultMap;
 9     public void setWorkQueue(Queue<Object> workQueue){
10         this.workQueue= workQueue;
11     }
12     public void setResultMap(Map<String ,Object> resultMap){
13         this.resultMap=resultMap;
14     }
15     //子任务处理的逻辑，在子类中实现具体逻辑
16     public Object handle(Object input){
17         return input;
18     }
19     
20     @Override
21     public void run() {
22         while(true){
23             //获取子任务
24             Object input= workQueue.poll();
25             if(input==null){
26                 break;
27             }
28             //处理子任务
29             Object re = handle(input);
30             resultMap.put(Integer.toString(input.hashCode()), re);
31         }
32     }
33 }

　　以上两段代码已经展示了 Master-Worker 框架的全貌。应用程序中通过重载Worker.handle()方法实现应用层逻辑。

注意：Master-Worker 模式是一种将串行任务并行化的方法，被分解的子任务在系统中可以被并行处理。同时，如果有需要，Master 进程不需要等待所有子任务都完成计算，就可以根据已有的部分结果集计算最终结果。

　　现应用这个 Master-Worker 框架，实现一个计算立方和的应用，并计算 1 ~~ 100 的立方和，即 1³ + 2³  + 3³ + ... + 100³。

　　计算任务可被分解为 100 个子任务，每个子任务仅用于计算单独的立方和。Master 产生固定个数的 Worker 来处理所有这些子任务。Worker 不断地从任务集合中取得这些计算立方和的子任务，并将计算结果返回给 Master。Master 负责将所有 Worker 的任务结果进行累加，从而产生最终的立方和。在整个计算过程中，Master 与 Worker 的运行也是完全异步的，Master 不必等到所有的 Worker 都执行完成后，就可以进行求和操作。即，Master 在获得部分子任务结果集时，就已经可以开始对最终结果进行计算，从而进一步提高系统的并行度和吞吐量。具体的任务分解如下图所示：

　　

3.子任务 PlusWork 源码实现

　　计算任务被划分成100个子任务，每个任务仅仅用于计算单独的立方和，对应的 PlusWork 源码如下：

public class PlusWorker extends Worker { //求立方和
    @Override
    public Object handle(Object input) {
        int i = (Integer)input;
        return i * i * i;
    }
}

4. 进行计算的 Main 函数

　　运行的调用函数如下。在主函数中首先通过Master类创建5个Worker工作进程和Worker工作实例PlusWorker。在提交了100个子任务后，边开始子任务的计算。这些子任务中由这5个进程共同完成。Master不用等待所有Worker计算完成才开始汇总，而是子任务在计算的过程中，Master就开始汇总了。

 1 import java.util.Map;
 2 import java.util.Set;
 3 
 4 public class Application {
 5     public static void main(String[] args) {
 6         //固定使用5个Workde
 7         Master master = new Master(new PlusWorker(), 5);
 8         for(int i=1; i<=100; i++) //提交100个子任务
 9             master.submit(i);
10         master.execute(); //开始计算
11         int re = 0;  //最终计算结果保存在此
12         Map<String, Object> resultMap = master.getResultMap();
13         
14         while(true) {//不需要等待所有Worker都执行完成，即可开始计算最终结果
15             Set<String> keys = resultMap.keySet();  //开始计算最终结果
16             String key = null;
17             for(String k : keys) {
18                 key = k;
19                 break;
20             }
21             Integer i = null;
22             if(key != null)
23                 i = (Integer)resultMap.get(key);
24             if(i != null)
25                 re += i; //最终结果
26             if(key != null)
27                 resultMap.remove(key); //移除已被计算过的项目
28             if(master.isComplete() && resultMap.size()==0)
29                 break;
30         }
31         System.out.println(re);
32     }
33 }

运行结果：

25502500

 总结：

　　重要的事情说三遍，Master-Worker 模式是一种将串行任务并行化的方案，被分解的子任务在系统中可以被并行处理，同时，如果有需要，Master进程不需要等待所有子任务都完成计算，就可以根据已有的部分结果集计算最终结果集。