线程学习--Master-Worker模式

Posted 2020-09-22

　master-worker模式是一种并行计算模式，分为master进程和worker进程两个部分，master是担任总管角色，worker才是执行具体任务的地方。

　　总体流程应该是这样的：

 

package masterworker;

import java.util.Random;

public class Main {
    
    public static void main(String[] args) {
        
        Master master=new Master(new Worker(), 20);
        
        Random r=new Random();
        for(int i=1;i<=100;i++){
            Task t=new Task();
            t.setId(i);
            t.setName("任务"+i);
            t.setPrice(r.nextInt(1000));
            master.submit(t);
        }
        
        master.execute();
        
        long start = System.currentTimeMillis();
        while(true){
            if(master.isComplete()){
                long end = System.currentTimeMillis() - start;
                int ret = master.getResult();
                System.out.println("最终结果：" + ret + "， 执行耗时：" + end);
                break;
            }
        }
        
    }
}

View Code

package masterworker;


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

public class Master {

    //1 应该有一个承装任务的集合
    private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
    
    //2 使用HashMap去承装所有的worker对象
    private HashMap<String, Thread> workers = new HashMap<String, Thread>();
    
    //3 使用一个容器承装每一个worker并非执行任务的结果集
    private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
    
    //4 构造方法
    public Master(Worker worker, int workerCount){
        // 每一个worker对象都需要有Master的引用 workQueue用于任务的领取，resultMap用于任务的提交
        worker.setWorkerQueue(this.workQueue);
        worker.setResultMap(this.resultMap);
        for(int i = 0 ; i < workerCount; i++){
            //key表示每一个worker的名字, value表示线程执行对象
            workers.put("子节点" + Integer.toString(i), new Thread(worker));
        }
    }
    
    //5 提交方法
    public void submit(Task task){
        this.workQueue.add(task);
    }
    
    //6 需要有一个执行的方法（启动应用程序 让所有的worker工作）
    public void execute(){
        for(Map.Entry<String, Thread> me : workers.entrySet()){
            me.getValue().start();
        }
    }

    //8 判断线程是否执行完毕
    public boolean isComplete() {
        for(Map.Entry<String, Thread> me : workers.entrySet()){
            if(me.getValue().getState() != Thread.State.TERMINATED){
                return false;
            }
        }        
        return true;
    }

    //9 返回结果集数据
    public int getResult() {
        int ret = 0;
        for(Map.Entry<String, Object> me : resultMap.entrySet()){
            //汇总的逻辑..
            ret += (Integer)me.getValue();
        }
        return ret;
    }
    
    
    
    
    
    
    
    
}

View Code

package masterworker;

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


public class Worker implements Runnable {

    private ConcurrentLinkedQueue<Task> workQueue;
    private ConcurrentHashMap<String, Object> resultMap;
    
    public void setWorkerQueue(ConcurrentLinkedQueue<Task> workQueue) {
        this.workQueue = workQueue;
    }

    public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
        this.resultMap = resultMap;
    }
    
    
    @Override
    public void run() {
        while(true){
            Task input = this.workQueue.poll();
            if(input == null) break;
            //真正的去做业务处理
            Object output =handle(input);
            this.resultMap.put(Integer.toString(input.getId()), output);
        }
    }
    
    public static Object handle(Task input) {
        Object output = null;
        try {
            //表示处理task任务的耗时，可能是数据的加工，也可能是操作数据库...
            Thread.sleep(500);
            output = input.getPrice();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return output;
    }

}

View Code

package masterworker;

public class Task {
    private int id ;
    private String name;
    private int price;
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public int getPrice() {
        return price;
    }
    public void setPrice(int price) {
        this.price = price;
    }
    
    
}

View Code