Master-Worker設計模式介紹

Master-Worker模式是常用的並行設計模式。核心思想是，系統由兩個角色組成，Master和Worker，Master負責接收和分配任務，Worker負責處理子任務。任務處理過程中，Master還負責監督任務進展和Worker的健康狀態；Master將接收Client提交的任務，並將任務的進展匯總反饋給Client。各角色關系如下圖 

Master-Worker模式滿足於可以將大任務划分為小任務的場景，是一種分而治之的設計理念。通過多線程或者多進程多機器的模式，可以將小任務處理分發給更多的CPU處理，降低單個CPU的計算量，通過並發/並行提高任務的完成速度，提高系統的性能。

具體細節如上圖，Master對任務進行切分，並放入任務隊列；然后，觸發Worker處理任務。實際操作中，任務的分配有多種形式，如Master主動拉起Workder進程池或線程池，並將任務分配給Worker；或者由Worker主動領取任務，這樣的Worker一般是常駐進程；還有一種解耦的方式，即Master指做任務的接收、切分和結果統計，指定Worker的數量和性能指標，但不參與Worker的實際管理，而是交由第三方調度監控和調度Worker。

 

代碼實現Master-Worker模式:

Master代碼:

package com.hjf.master_worker;

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

/**
 * Master
 * @author huangjianfei
 */
public class Master
{
    //1:應該有一個承載任務的集合
    private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
    
    //2:使用hashmap去承載所有的worker對象 ThreadName------Worker
    private HashMap<String,Thread> workers = new HashMap<>();
    
    //3:使用一個容器承載每一個worker並行執行任務的結果集
    private ConcurrentHashMap<String,Object> resultMap = new ConcurrentHashMap<String, Object>();
    
    //4:構造方法
    public Master(Worker worker,int workerCount){
        //在worker中添加兩個引用  workQueue用於任務的領取  resultMap用於任務的提交
        worker.setWorkerQueue(this.workQueue);
        worker.setResultMap(this.resultMap);
        
        for (int i = 0; i < workerCount; i++)
        {
            workers.put("子節點 "+i, new Thread(worker));
        }
    }
    
    //5:提交方法
    public void submit(Task task){
        workQueue.add(task);
    }
    
    //6:需要有一個執行的方法(啟動應用程序 讓所有的worker工作)
    public void execute(){
        //遍歷workers 分別去執行每一個worker
        for (Map.Entry<String,Thread> me: workers.entrySet())
        {
            me.getValue().start();
        }
    }

    /**
     * 判斷所有的worker是否執行完畢
     */
    public boolean isCompleted()
    {
        //遍歷所有的worker 只要有一個沒有停止 那么就代表沒有結束
        for (Map.Entry<String,Thread> me: workers.entrySet())
        {
            if(me.getValue().getState() != Thread.State.TERMINATED){
                return false;
            }
        }
        return true;
    }
    
    /**
     * 計算最終的結果集
     * @return
     */
    public int getResult(){
        int result = 0;
        for (Map.Entry<String,Object> me : resultMap.entrySet())
        {
            result += (Integer)me.getValue();
        }
        return result;
    }
}

 

Worker代碼實現:

package com.hjf.master_worker;

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

/**
 * Worker
 * @author huangjianfei
 */
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(null == input) break;
            //真正的去做業務處理
            Object outPut = handle(input);
            //存放任務的結果 
            this.resultMap.put(String.valueOf(input.getId()), outPut);
        }  
    }

    //單獨抽出來 給子類重寫，更加靈活
    public Object handle(Task input){
        return null;
    }
    
    
    /**
     * 處理業務 應該抽象出來 子類去具體實現業務邏輯
     * @param input
     */
//    private Object handle(Task input)
//    {
//        Object outPut = null;
//        if(null == input) return null;
//        try
//        {
//            //表示處理task任務的耗時，可能是數據的加工，也可能是操作數據庫
//            Thread.sleep(5000);
//            //模擬真實的業務場景
//            outPut = input.getPrice();
//        } catch (InterruptedException e)
//        {
//            e.printStackTrace();
//        }
//        return outPut;
//    }
    
}

 

Task代碼實現:

package com.hjf.master_worker;
/**
 * 任務
 * @author huangjianfei
 */
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;
    }
    
}

 

Worker子類，在以后的開發中可以按照自己的需求去設計相關的Worker的子類:

package com.hjf.master_worker;

public class MyWorker1 extends Worker
{
    @Override 
    public Object handle(Task input)
    {
        Object outPut = null;
        if(null == input) return null;
        try
        {
            //表示處理task任務的耗時，可能是數據的加工，也可能是操作數據庫
            Thread.sleep(5000);
            //模擬真實的業務場景
            outPut = input.getPrice();
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        return outPut;
    }
}

 

Main測試類代碼:

package com.hjf.master_worker;

import java.util.Random;
/**
 * 主線程測試類
 * @author huangjianfei
 */
public class Main
{
    public static void main(String[] args)
    {
        System.out.println("我的機器可用的Processor數量:"+Runtime.getRuntime().availableProcessors());
        // 使用worker子類實現具體的業務,更加靈活
        Master master = new Master(new MyWorker1(), Runtime.getRuntime().availableProcessors()); 
        Random r = new Random();
        //提交100個任務
        for (int i = 0; i <= 100; i++)
        {
            Task t = new Task();
            t.setId(i);
            t.setName("任務 "+i);
            t.setPrice(r.nextInt(1000)); 
            master.submit(t);
        }
        
        //執行所有的worker
        master.execute();
        
        long start = System.currentTimeMillis();//記錄時間
        
        
        while(true){
            //全部的worker執行結束的時候去計算最后的結果
            if(master.isCompleted()){
                long end = System.currentTimeMillis() - start;//計算耗時
                //計算結果集
                int result = master.getResult();
                System.out.println("執行最終結果: "+result + " 執行耗時 "+end);
                break;
            }
        }
        
    }
    
}