并發程式設計：Master-Worker模式

一、Master-Worker模式

Master-Worker的工作機制就是Master接收到了一個任務,對任務進行拆分,并且配置設定給各個Worker,讓各個Worker之間各自運作各自的子任務,最後Worker們的傳回結果交給Master進行彙總并且最後傳回給任務的發起方.

Master-Worker是一種并行模式,Master是主要程序,Master中有維護着一個Worker程序隊列.Master把一個大的而且複雜的業務拆分成若幹小的業務,隻要是互不影響的都可以分而治之互相獨立.可以通過多線程或多程序甚至多機聯合計算,把拆分後的小業務交給更多的CPU或機器處理,通過并發/并行的方式提高整體業務的運算速度,壓榨系統性能來提高效率.

二、Master-Worker模式Demo

 Master角色

package ConcurrentProgramming.middle.part3.MasterWorker;


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

/**
 * @Author: zdj
 * @Description: Master角色 接收和配置設定任務
 * @Date: 2019年03月15日 14:00
 */
public class Master {

    //定義一個容器 用來儲存任務
    private ConcurrentLinkedQueue<Task> taskQueue = new ConcurrentLinkedQueue<Task>();

    //定義一個存放Worker的容器
    private HashMap<String,Thread> workers = new HashMap<String, Thread>();

    //定義一個容器存放執行結果
    private ConcurrentHashMap<String,Object> result = new ConcurrentHashMap<String, Object>();

    public Master(Worker worker,int workerCount){
        worker.setTaskQueue(this.taskQueue);
        worker.setResult(this.result);
        for (int i = 0; i <workerCount; i++){
            this.workers.put(Integer.toString(i),new Thread(worker));
        }
    }

    //送出任務的方法 将任務裝到任務隊列
    public void submit(Task task){
        this.taskQueue.add(task);
    }

    //啟動線程去執行任務
    public void execute(){
        for (Map.Entry<String,Thread> me : workers.entrySet()){
            me.getValue().start();
        }
    }

    //判斷是否運作結束
    public boolean isComplete(){
        for (Map.Entry<String,Thread> me : workers.entrySet()){
            if (me.getValue().getState() != Thread.State.TERMINATED){
                return false;
            }
        }
        return true;
    }

    //擷取計算結果
    public String getResult(){
        String s = "";
        for (Map.Entry me:result.entrySet()){
            s += (String) me.getKey();
        }
        return s;
    }

}
           

worker角色

package ConcurrentProgramming.middle.part3.MasterWorker;

import lombok.Data;

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

/**
 * @Author: zdj
 * @Description: Worker角色 負責執行任務
 * @Date: 2019年03月15日 14:06
 */
@Data
public class Worker implements Runnable{

    //任務隊列引用
    private ConcurrentLinkedQueue<Task> taskQueue;

    //結果引用
    private ConcurrentHashMap<String,Object> result;

    public void run() {
        while (true){
            //從任務隊列中取任務
            Task task = this.taskQueue.poll();
            if (task == null){
                break;
            }
            //處理
            Object output = handle(task);
            //将結果放到結果隊列
            this.result.put(task.getId().toString(),output);
        }
    }
    
    //模拟處理任務
    private Object handle(Task task){
        Object output = null;
        try {
            //模拟處理耗時任務
            Thread.sleep(500);
            output = task.getName();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return output;
    }
}
           

任務Task

package ConcurrentProgramming.middle.part3.MasterWorker;

import lombok.Data;

/**
 * @Author: zdj
 * @Description: 任務
 * @Date: 2019年03月15日 14:01
 */
@Data
public class Task {
    private String id;
    private String name;
}
           

測試Main

package ConcurrentProgramming.middle.part3.MasterWorker;

/**
 * @Author: zdj
 * @Description: Master-Worker模式是常用的并行計算模式 Master負責接收和配置設定任務，Worker負責處理任務
 * @Date: 2019年03月15日 14:36
 */
public class Main {
    public static void main(String[] args) {

        //建立20個woker去執行任務
        Master master = new Master(new Worker(),20);

        //建立100個任務
        for (int i = 1; i <= 100 ;i++){
            Task task = new Task();
            task.setId("id" + Integer.toString(i));
            task.setName("name" + Integer.toString(i));
            master.submit(task);
        }
        master.execute();
        //記錄下開始時間
        long start = System.currentTimeMillis();

        while (true){
            if (master.isComplete()){
                long end = System.currentTimeMillis()-start;
                String result = master.getResult();
                System.out.println("結果為" + result + "\n執行時間為:" + end);
                break;
            }
        }
    }
}
           

運作結果：

參考連結：https://blog.csdn.net/unscdf117/article/details/78888814