Master-Worker模式

2020-07-13 10:25:25 浏览数 (1)

原文地址：http://www.wangqi94.com/to/master/blog?uuid=80

前言

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

Master-Worker工作模式示意图.pngMaster-Worker工作模式示意图.png

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

Master-Worker模式结构

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

Master-Worker模式结构图.pngMaster-Worker模式结构图.png

其中Master-Worker模式的主要参与者作用如下：

代码语言：txt复制
Worker：用于实际处理一个任务
Master：任务的分配和最终结果的合成
Main：启动系统，调度开启Master
Master-Worker模式的代码实现

基于上述的设计思路，这里给我们给出一个简易的Master-Worker框架。其中Master部分的实现如下：

代码语言：txt复制
/**
 * @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部分的代码实现：

代码语言：txt复制
/**
 * @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。 任务分解如下图所示：

任务分解.png任务分解.png

分析：计算任务将被分解为100个子任务，每个子任务仅用于计算单独的立方和。Master产生固定个数的Worker，来梳理所有这些子任务。Worker不断地从任务集合中取得这些计算立方和的子任务，并将计算结果返回给Master。最后，Master负责将所有Worker的任务结果进行累加，从而产生最终的立方和。

在整个计算过程中，Master和Worker的运行也是完全异步的，Master不必等到所有的Worker都执行完成后，就可以进行求和操作。即，Master在获得部分子任务结构机时，就已经开始对最终结果进行计算，从而进一步提高系统的并行度和吞吐量。

Worker对象在应用层的代码实现如下：

代码语言：txt复制

/**

  • @Author: 王琦 <QQ.Eamil>1124602935@qq.com</QQ.Eamil>
  • @Date: 2019-5-3 0003 12:40
  • @Description: 一个具体的Worker : PlusWorker */ public class PlusWorker extends Worker {
代码语言：txt复制
// Worker, 求立方
代码语言：txt复制
@Override
代码语言：txt复制
public Object handle(Object input) {
代码语言：txt复制
    Integer n = (Integer) input;
代码语言：txt复制
    return n * n * n;
代码语言：txt复制
}

}

代码语言：txt复制
使用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 {
代码语言：txt复制
public static void main(String[] args) {
代码语言：txt复制
    StopWatch watch = new StopWatch("使用Master-Worker模式计算1~100的立方和任务开始....");
代码语言：txt复制
    watch.start();
代码语言：txt复制
    // 1~100求立方和的最终结果：result
代码语言：txt复制
    int result = 0;
代码语言：txt复制
    // 实例化一个Master ： 固定使用5个Worker， 并指定具体的Worker（即PlusWorker）
代码语言：txt复制
    Master master = new Master(new PlusWorker(), 5);
代码语言：txt复制
    for (int i=0; i<100; i  ){
代码语言：txt复制
        // 提交100个子任务
代码语言：txt复制
        master.submit(i);
代码语言：txt复制
    }
代码语言：txt复制
    // 开始计算
代码语言：txt复制
    master.execute();
代码语言：txt复制
    Map<String, Object> resultMap = master.getResultMap();
代码语言：txt复制
    while (resultMap.size() > 0 || !master.isComplete()){
代码语言：txt复制
        // 注意：我们这里不需要等待所有的Worker都执行完。
代码语言：txt复制
        Set<String> keys = resultMap.keySet();
代码语言：txt复制
        String key = null;
代码语言：txt复制
        for (String k : keys){
代码语言：txt复制
            key = k;
代码语言：txt复制
            break;
代码语言：txt复制
        }
代码语言：txt复制
        Integer i = null;
代码语言：txt复制
        if (key != null){
代码语言：txt复制
            i = (Integer) resultMap.getOrDefault(key, null);
代码语言：txt复制
        }
代码语言：txt复制
        // Master求和
代码语言：txt复制
        if (i != null){
代码语言：txt复制
            result  = i;
代码语言：txt复制
        }
代码语言：txt复制
        // 移除已经被计算过的项
代码语言：txt复制
        if (key != null){
代码语言：txt复制
            resultMap.remove(key);
代码语言：txt复制
        }
代码语言：txt复制
    }
代码语言：txt复制
    watch.stop();
代码语言：txt复制
    System.out.println("1~100立方和："  result);
代码语言：txt复制
    System.out.println("任务用时："  watch.getTotalTimeMillis());
代码语言：txt复制
}

}

代码语言：txt复制
控制台输出:
```java

1~100立方和：24502500

任务用时：217

代码语言：txt复制
总结

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

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

class com date public worker

0 人点赞