Master-Worker模式

前言

---

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：启动系统，调度开启Master

Master-Worker模式的代码实现

---

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

/**
 * @Author: 王琦 <QQ.Eamil>1124602935@qq.com</QQ.Eamil>
 * @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部分的代码实现：

/**
 * @Author: 王琦 <QQ.Eamil>1124602935@qq.com</QQ.Eamil>
 * @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, 求立方
   @Override
   public Object handle(Object input) {
       Integer n = (Integer) input;
       return n * n * n;
   }
}

使用Master-Worker框架进行计算的Main如下：

/**
 * @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());

    }
}

控制台输出:

1~100立方和：24502500
任务用时：217

总结

---

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

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

我的更多博文：https://www.relaxheart.cn/to/master/blog