Master-Worker模式
原创
原文地址：http://www.wangqi94.com/to/master/blog?uuid=80
前言
Master-Worker模式是常用的并行模式之一。它的核心思想是系统由两类进程协作工作：Master进程和Worker进程。Master进程负责接收和分配任务，Worker进程负责处理子任务。当个个进程将子任务处理完成后，将结果返回给Master进程，由Master进程做归纳和汇总，从而得到系统的最终结果，其处理过程图下图所示：
Master-Worker模式的好处，它能够将一个大任务分解成若干个小任务，并行执行，从而提高系统的吞吐量。而对于系统请求者Client来说，任务一旦提交，Master进程会分配任务并立即返回，不会等待系统全部处理完成后在返回，其处理进程是异步的,因此Client不会出现等待现象。
Master-Worker模式结构
Master-Worker模式的结构相对比较简单，我们下面将给出一个简明的实现方式。如下图所示，Master进程为主进程，它维护了一个Worker进程队列，子任务队列和子结果集。Worker进程队列中的Worker进程不停地从任务队列中提取要处理的子任务，并将子任务的处理结果写入结果集。
其中Master-Worker模式的主要参与者作用如下：
Worker：用于实际处理一个任务
Master：任务的分配和最终结果的合成
Main：启动系统，调度开启MasterMaster-Worker模式的代码实现
基于上述的设计思路，这里给我们给出一个简易的Master-Worker框架。其中Master部分的实现如下：
/**
* @Date: 2019-5-3 0003 12:16
* @Description: cn.relaxheart.master.worker.framework : Master
*/
public class Master {
// Master维护一个任务队列
protected Queue<Object> workQueue = new ConcurrentLinkedQueue<>();
// Worker进程队列
protected Map<String, Thread> threadMap = new HashMap<>();
// 子任务处理结果集
protected Map<String, Object> resultMap = new ConcurrentHashMap<>();
public Master(Worker worker, int countWorker){
worker.setWorkQueue(workQueue);
worker.setResultMap(resultMap);
for (int i=0; i<countWorker; i++){
threadMap.put(Integer.toString(i), new Thread(worker, Integer.toString(i)));
}
}
/**
* 是否所有子任务都结束
* @return
*/
public boolean isComplete(){
for (Map.Entry<String, Thread> entry:threadMap.entrySet()){
if (entry.getValue().getState() != Thread.State.TERMINATED){
return false;
}
}
return true;
}
/**
* 提交一个任务
* @param job
*/
public void submit(Object job){
workQueue.add(job);
}
public Map<String, Object> getResultMap(){
return resultMap;
}
/**
* 开始运行所有的Worker进程，进行处理
*/
public void execute(){
for (Map.Entry<String, Thread> entry:threadMap.entrySet()){
entry.getValue().start();
}
}
}Worker部分的代码实现：
/**
* @Date: 2019-5-3 0003 12:22
* @Description: cn.relaxheart.master.worker.framework : Worker
*/
public class Worker implements Runnable{
// 任务队列，用于取得子任务
protected Queue<Object> workQueue;
// zi任务处理结果集
protected Map<String, Object> resultMap;
public void setWorkQueue(Queue<Object> workQueue){
this.workQueue = workQueue;
}
public void setResultMap(Map<String, Object> resultMap) {
this.resultMap = resultMap;
}
/**
* 子任务处理Handle，在具体的子任务实现具体的逻辑
* @param input
* @return
*/
public Object handle(Object input){
return input;
}
@Override
public void run() {
while (true){
// 获取子任务
Object input = workQueue.poll();
if (input == null)
break;
Object result = handle(input);
// 将处理结果写入结果集
resultMap.put(Integer.toString(input.hashCode()), result);
}
}
}以上两段代码已经展示了Master-Worker框架的全貌。应用程序总通过重载Worker.handle()方法实现应用层逻辑。
下面应用我们自己设计的这个Master-Worker框架，实现一个计算立方和的应用，并计算1~100的立方和，即1^3 + 2^3 + ... + 100^3。 任务分解如下图所示：
分析：计算任务将被分解为100个子任务，每个子任务仅用于计算单独的立方和。Master产生固定个数的Worker，来梳理所有这些子任务。Worker不断地从任务集合中取得这些计算立方和的子任务，并将计算结果返回给Master。最后，Master负责将所有Worker的任务结果进行累加，从而产生最终的立方和。
在整个计算过程中，Master和Worker的运行也是完全异步的，Master不必等到所有的Worker都执行完成后，就可以进行求和操作。即，Master在获得部分子任务结构机时，就已经开始对最终结果进行计算，从而进一步提高系统的并行度和吞吐量。
Worker对象在应用层的代码实现如下：
/**
- @Author: 王琦 <QQ.Eamil>1124602935@qq.com</QQ.Eamil>
- @Date: 2019-5-3 0003 12:40
- @Description: 一个具体的Worker : PlusWorker */ public class PlusWorker extends Worker {
// Worker, 求立方@Overridepublic Object handle(Object input) { Integer n = (Integer) input; return n * n * n;}}
使用Master-Worker框架进行计算的Main如下：
```java/**
- @Author: 王琦 <QQ.Eamil>1124602935@qq.com</QQ.Eamil>
- @Date: 2019-5-3 0003 12:42
- @Description: Master-Worker求立方和的Main */ public class PlusMasterWorkerMain {
public static void main(String[] args) { StopWatch watch = new StopWatch("使用Master-Worker模式计算1~100的立方和任务开始...."); watch.start(); // 1~100求立方和的最终结果：result int result = 0; // 实例化一个Master ： 固定使用5个Worker， 并指定具体的Worker（即PlusWorker） Master master = new Master(new PlusWorker(), 5); for (int i=0; i<100; i++){ // 提交100个子任务 master.submit(i); } // 开始计算 master.execute(); Map<String, Object> resultMap = master.getResultMap(); while (resultMap.size() > 0 || !master.isComplete()){ // 注意：我们这里不需要等待所有的Worker都执行完。 Set<String> keys = resultMap.keySet(); String key = null; for (String k : keys){ key = k; break; } Integer i = null; if (key != null){ i = (Integer) resultMap.getOrDefault(key, null); } // Master求和 if (i != null){ result += i; } // 移除已经被计算过的项 if (key != null){ resultMap.remove(key); } } watch.stop(); System.out.println("1~100立方和："+ result); System.out.println("任务用时："+ watch.getTotalTimeMillis());}}
控制台输出:
```java1~100立方和：24502500
任务用时：217
总结
（1）Master-Worker模式是一种使用多线程进行数据处理的结构。多个Worker进行协作处理用户请求，Master金城负责维护Worker进程，并整合最终的处理结果。
（2）Master-Worker模式是一种将串行任务并行化的方法，被分解额子任务在系统中可以被并行处理。同时，如果有需求，Master进程不需要等待所有子任务都完成才做最终的整合，就可以根据已有的部分结果集整合做种的结果。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
腾讯云开发者
