Master-worker模式

master-worker模式是常用的并行模式。就像软件公司的工作模式一样,客户将需求提给项目经理,项目经理将需求分解,然后分配给各个开发人员,开发人员开发完成后将结果反馈给项目经理,由项目经理向客户交付。在这种工作模式之中,项目经理扮演的角色就是Master,开发人员的角色就是worker。master-worker模式的核心思想是,系统有两个进程协作工作:Master进程,负责接收和分配任务;Worker进程,负责处理子任务。当Worker进程将子任务处理完成后,结果返回给Master进程,由Master进程做归纳汇总,最后得到最终的结果。

下面给出该模式的一个简单实现,可以到我的github中下载源码

master类代码:此处master类实现了三个方法,分别用来提交任务(submitTask),执行任务(execute),判断任务是否完成(isComplate),以及返回结果(getResult);

package com.fullstacker.study.designpattern.MasterWorkers;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * master类
 *
 * @author xingguishuai
 * @create 2017-04-13-17:21
 **/
public class Master {
    /**
     *用来保存任务,由于会涉及多线程共同处理任务所以使用concurrent包
     */
    private ConcurrentLinkedQueue<Task> tasks = new ConcurrentLinkedQueue<Task>();
    /**
     * 保存结果 由于会涉及多线程共同处理任务所以使用concurrent包
     */
    private Map<String,Object> resultMap = new ConcurrentHashMap<String, Object>();
    /**
     * 用来保存workers
     */
    private HashMap<String,Thread> workers = new HashMap<String, Thread>();
    /**
    * 

功能描述:构造函数,指定work和work数量


    * @return
    * @param
    * @author xingguishuai
    * @Date 2017-04-13 17:50
    * @since 1.0
    */
    public Master(Worker worker,int workerCount){
        worker.setResultMap(resultMap);
        worker.setTasks(tasks);
        for (int i = 0;i < workerCount;i++){
            String workName = "workerName-" + i;
            workers.put(workName,new Thread(worker,workName));
        }
    }
    /**
    * <p>功能描述:提交任务</p>
    * @return
    * @param
    * @author xingguishuai
    * @Date 2017-04-13 18:02
    * @since 1.0
    */
    public void submitTask(Task task){
        tasks.add(task);
    }
    /**
    * <p>功能描述:master调用work执行任务</p>
    * @return
    * @param
    * @author xingguishuai
    * @Date 2017-04-13 18:01
    * @since 1.0
    */
    public void execute(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            entry.getValue().start();
        }
    }
    /**
    * <p>功能描述:判断任务是否执行完</p>
    * @return
    * @param
    * @author xingguishuai
    * @Date 2017-04-13 18:06
    * @since 1.0
    */
    public boolean isComplate(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            //如果有一个线程的状态不为TERMINATED,即可标志任务未完成
            if(!entry.getValue().getState().equals(Thread.State.TERMINATED)){
                return  false;
            }
        }
        return  true;
    }
    /**
    * <p>功能描述:返回结果集</p>
    * @return
    * @param
    * @author xingguishuai
    * @Date 2017-04-13 18:26
    * @since 1.0
    */
    public Map<String, Object> getResult(){
        return  resultMap;
    }
}

worker类,Woker类为一个实现了Runnable接口的抽象类,task属性用来记录Master中的任务,resultMap属性用来保存处理结果。抽象方法handle()用来实际处理任务,在各子类中实现该方法,方便扩展。

package com.fullstacker.study.designpattern.MasterWorkers;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * worker类
 *
 * @author xingguishuai
 * @create 2017-04-13-17:21
 **/
public abstract class Worker implements  Runnable{
    /**
     *用来获取master中的任务
     */
    private ConcurrentLinkedQueue<Task> tasks ;
    /**
     * 保存结果
     */
    private Map<String,Object> resultMap ;
    @Override
    public void run() {
        //只要存在任务就一直循环取出任务并执行
        while (true){
            //取出一个任务
            Task task = tasks.poll();
            //任务执行完毕
            if(null == task){
                break;
            }
            Object result = handle(task);
            //保存结果
            resultMap.put(Thread.currentThread().getName()+":"+task.hashCode(),result);
        }
    }

    public abstract Object handle(Task task);

    public ConcurrentLinkedQueue<Task> getTasks() {
        return tasks;
    }

    public void setTasks(ConcurrentLinkedQueue<Task> tasks) {
        this.tasks = tasks;
    }

    public Map<String, Object> getResultMap() {
        return resultMap;
    }

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

task类

package com.fullstacker.study.designpattern.MasterWorkers;

/**
 * task类
 *
 * @author xingguishuai
 * @create 2017-04-13-17:27
 **/
public interface Task {

    public abstract Object doTask();

}

下面是一个Master-worker模式的使用示例:
利用该模式计算学生的总成绩.

GradeTask

package com.fullstacker.study.designpattern.MasterWorkers.example;

import com.fullstacker.study.designpattern.MasterWorkers.Task;

/**
 * 计算学生成绩任务
 *
 * @author xingguishuai
 * @create 2017-04-13-18:30
 **/
public class GradeTask implements Task {
    /**
     * 学生ID
     */
    private Long sutudentId;
    /**
     * 数学成绩
     */
    private float mathGrade;
    /**
     * 语文成绩
     */
    private float chineseGrade;

    public Long getSutudentId() {
        return sutudentId;
    }

    public void setSutudentId(Long sutudentId) {
        this.sutudentId = sutudentId;
    }

    public float getMathGrade() {
        return mathGrade;
    }

    public void setMathGrade(float mathGrade) {
        this.mathGrade = mathGrade;
    }

    public float getChineseGrade() {
        return chineseGrade;
    }

    public void setChineseGrade(float chineseGrade) {
        this.chineseGrade = chineseGrade;
    }

    @Override
    public Object doTask() {
        return chineseGrade + mathGrade;
    }
}

GradeWorker

package com.fullstacker.study.designpattern.MasterWorkers.example;

import com.fullstacker.study.designpattern.MasterWorkers.Task;
import com.fullstacker.study.designpattern.MasterWorkers.Worker;

/**
 * 计算成绩的worker
 *
 * @author xingguishuai
 * @create 2017-04-13-18:34
 **/
public class GradeWorker extends Worker {
    @Override
    public Object handle(Task task) {
        return task.doTask();
    }
}

在main方法中调用

package com.fullstacker.study.designpattern.MasterWorkers.example;

import com.fullstacker.study.designpattern.MasterWorkers.Master;
import com.fullstacker.study.designpattern.MasterWorkers.Worker;

import java.util.Map;
import java.util.Random;

import static javafx.scene.input.KeyCode.R;

/**
 * 测试master worker模式
 *
 * @author xingguishuai
 * @create 2017-04-13-18:39
 **/
public class Test {
    public static void main(String[] args){
        Worker worker = new GradeWorker();

        Master master = new Master(worker,Runtime.getRuntime().availableProcessors());
        Random rundom = new Random();
        for (int i = 0; i < 4; i++) {
            GradeTask gradeTask = new GradeTask();
            gradeTask.setChineseGrade(rundom.nextFloat()*100);
            gradeTask.setMathGrade(rundom.nextFloat()*100);
            master.submitTask(gradeTask);
        }
        master.execute();
        while (true){
            if(master.isComplate()){
                Map<String, Object> result = master.getResult();
                for (Map.Entry<String, Object> me : result.entrySet()) {
                    System.out.println(me.getKey()+":::"+me.getValue());
                }
                break;
            }
        }
    }
}

创作不易,转载请注明文章来源为www.full-satcker.com:邢桂帅 » Master-worker模式

赞 (1)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址