Java并發(fā)案例05---Master-Worker模式

Master-Worker 模式是常用的并行計算模式。它的核心思想是系統(tǒng)由兩類進程協(xié)同工作跳昼，Master和Worker進程。Master負責接收和分配任務，Worker負責處理子任務闰非。當各個Worker子進程處理完畢后，會將結果返回給Master峭范，由Master做歸納和小結财松。其好處是能夠將一個大任務分解成若干個小任務，并行執(zhí)行纱控，從而提高系統(tǒng)的吞吐量辆毡。

Master-Worker模式結構圖

結構圖1

結構圖2

代碼架構：

image

Master 負責接收和分配任務

Worker 負責處理子任務

Main 主函數(shù)，啟動類

Master類

public class Master {

    //1 有一個盛放任務的容器  1.任務不需要阻塞   2.性能高   3.線程安全,每一個worker來這里拿100個task甜害，所以要考慮線程安全舶掖。
    private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
    
    //2 需要有一個盛放worker的集合    1.不存在多線程爭搶場景
    private HashMap<String, Thread> workers = new HashMap<String, Thread>();
    
    //3 需要有一個盛放每一個worker執(zhí)行任務的結果集合   結果集，多個worker并發(fā)回寫操作
    private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
    
    //4 構造方法
    public Master(Worker worker , int workerCount){
        worker.setWorkQueue(this.workQueue);
        worker.setResultMap(this.resultMap);
        
        for(int i = 0; i < workerCount; i ++){
            this.workers.put(Integer.toString(i), new Thread(worker));
        }
        
    }
    
    //5 需要一個提交任務的方法
    public void submit(Task task){
        this.workQueue.add(task);
    }
    
    //6 需要有一個執(zhí)行的方法尔店，啟動所有的worker方法去執(zhí)行任務
    public void execute(){
        for(Map.Entry<String, Thread> me : workers.entrySet()){
            me.getValue().start();
        }
    }

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

    //8 計算結果方法
    public int getResult() {
        int priceResult = 0;
        for(Map.Entry<String, Object> me : resultMap.entrySet()){
            priceResult += (Integer)me.getValue();
        }
        return priceResult;
    }
    
}

需要說明的幾點：
1.存放任務的容器眨攘，需要一個高并發(fā)，線程安全闹获，性能好的容器期犬，因為每一個Worker都回來這里取任務執(zhí)行，存在線程安全問題避诽，因此選擇ConcurrentLinkedQueue龟虎。
2.存放結果的容器，同樣沙庐，worker也持有他的引用鲤妥，每一個worker做完事情后，會把結果寫入這個map中拱雏，存在線程安全問題棉安，所以采用ConcurrentHashMap。
3.存放worker的容器沒有線程并發(fā)訪問铸抑，只在Master類中進行了遍歷贡耽，不存在線程安全問題。

Worker類

public class Worker implements Runnable {

    private ConcurrentLinkedQueue<Task> workQueue;
    private ConcurrentHashMap<String, Object> resultMap;
    
    public void setWorkQueue(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);
            System.out.println("當前線程"+Thread.currentThread().getName()+"  計算完畢"+ input);
            this.resultMap.put(Integer.toString(input.getId()), output);
        }
    }

    private Object handle(Task input) {
        Object output = null;
        try {
            //處理任務的耗時。蒲赂。 比如說進行操作數(shù)據(jù)庫阱冶。。滥嘴。
            Thread.sleep(500);
            output = input.getPrice();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return output;
    }

}

Worker類需要實現(xiàn)Runnable接口

任務類

public class Task {

    private int id;
    private int price ;
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public int getPrice() {
        return price;
    }
    public void setPrice(int price) {
        this.price = price;
    }
    @Override
    public String toString() {
        return "Task [id=" + id + ", price=" + price + "]";
    } 
    
}

任務類比較簡單木蹬，不再敖述。

接下來是啟動類

public class Main {

    public static void main(String[] args) {
        /**20個worker*/
        Master master = new Master(new Worker(), 20);
        
        Random r = new Random();
        //100 個任務
        for(int i = 1; i <= 1000; i++){
            Task t = new Task();
            t.setId(i);
            t.setPrice(r.nextInt(1000));
            //添加任務
            master.submit(t);
        }
        //執(zhí)行任務
        master.execute();
        long start = System.currentTimeMillis();
        
        while(true){
            if(master.isComplete()){
                long end = System.currentTimeMillis() - start;
                int priceResult = master.getResult();
                System.out.println("最終結果：" + priceResult + ", 執(zhí)行時間：" + end);
                break;
            }
        }
        
    }
}

運行結果如下