多線程學習之路-學習master-worker設計模式

時間 2019-11-10

原文原文鏈接

前端時間主要精力在Java的遷移，就是把,以前項目的.net的代碼遷移成Java代碼，先完成的差很少了，主要考慮服務的優化，想起了Java的中master-worker模式，這個比較適用於多個計算的同時進行的優化，在網上找了一些學習的視頻，作了下面的一個練習，這個練習master-worker的練習的例子，不推薦實際項目這麼寫；前端

跟以前同樣先寫一下思路；java

首先申明Master和Worder兩個工具類，先說一下Master的實現思路dom

1.Master是負責調度的一個工具類，裏面確定應該有裝載任務的的集合，這裏使用了ConcurrentLinkedQueue;ide

2.Master是還須要負責給worker分配任務，這裏使用了HashMap來裝載worker對象函數

3.Master還須要負責對每個worker完成的結果彙總，由於每一個worker都會操做這個對象，因此定義了ConcurrentHashMap；工具

4.在Master的構造函數中把worker對象添加Master的引用，用於任務的領取和執行結果的提交；學習

5.後面分別是啓動方法、提交方法、收集執行結果的方法，詳細能夠看一下代碼；測試

package MasterWorkerTest;

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<>();

    // 4.構造方法
    public Master(Worker worker, int workerCount) {

        // 每一個worker對象都須要有master的引用workerQueue用於任務的領取，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();
        }
    }

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

    // 返回結果集數
    public int getReult() {
        int ret = 0;
        for (Map.Entry<String, Object> me : resultMap.entrySet()) {
            ret += (Integer) me.getValue();
        }

        return ret;
    }

}

下面的worker類優化

worker類其實就是實現了一個Runable接口，模擬業務邏輯操做；this

package MasterWorkerTest;


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() {
        // TODO Auto-generated method stub
        while (true) {
            Task input = this.workQueue.poll();
            if (input == null) {
                break;
            }
            // 作業務處理····
            Object outPut = Handle(input);
            this.resultMap.put(Integer.toString(input.getId()), outPut);
        }
    }

    // 業務邏輯處理
    private Object Handle(Task inPut) {
        Object outPut = null;
        // 模擬處理業務的耗時
        try {
            Thread.sleep(500);
            outPut = inPut.getPrice();

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        return outPut;
    }

}

下面是mian類，用於測試剛纔寫的代碼

package MasterWorkerTest;

import java.util.Random;

public class Main {

    public static void main(String[] args) {

        Random r = new Random();
        Master master = new Master(new Worker(), 10);
        for (int i = 1; i < 101; i++) {
            Task task = new Task();
            task.setId(i);
            task.setName("任務:" + i);
            task.setPrice(r.nextInt(1000));
            master.submit(task);
        }

        master.execute();

        long start = System.currentTimeMillis();
        while (true) {
            if (master.isComplete()) {
                long end = System.currentTimeMillis() - start;
                int ret = master.getReult();
                System.out.println("最終結果:" + ret + " 耗時:" + end);
                break;
            }

        }
    }

}

漏了一個實體類，這個是測試的返回實體，如今項目建議寫成泛型；

package MasterWorkerTest;

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;
    }
}