2. 程式人生 > >多執行緒並行設計模式之Master-Worker模式

多執行緒並行設計模式之Master-Worker模式

阿新 • • 發佈：2019-02-06

Master-Worker模式的核心思想是在於Master程序和Worker程序各自分擔各自的任務,協同完成資訊處理的模式 
使用一個圖例來簡單描述一下: 
Master-Worker工作原理 
如圖所示Master-Worker的工作機制就是Master接收到了一個任務,對任務進行拆分,並且分配給各個Worker,讓各個Worker之間各自執行各自的子任務,最後Worker們的返回結果交給Master進行彙總並且最後返回給任務的發起方.

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

這樣做的好處就是,在某些業務場景下,尤其是業務比較複雜而且資料量較大的情況下,例如財務賬單結算和生成.一個賬單需要有很多關聯計算而且條件和引數眾多的時候,如果把所有的業務都放在一個任務中,效率是比較低的,資料量大的情況下往往耗時巨大,少則幾十分鐘,誇張的有整整一宿.這個在圓通速遞的羅漢財務系統中體現尤其嚴重..不信可以問問許雙芳.

具體的實現細節是Master任務切分 –> 交給任務佇列 –> Worker處理任務 
Master可以建立Worker執行緒池,分發任務給Worker.也可以只負責任務的接收和拆分,單不負責Worker的管理,通過其他第三方工具來負責Worker的監控和排程,這樣對於解耦方面比較有利.

OK 有了思想之後 剩下的就是通過程式碼來簡單實現一下

先建立Master和Worker 
Master:

package com.unsc.Master_Worker;

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

/**
 * Created by DELL on 2017/12/25.
 * Master-Worker中的 Master 類
 * @author
 
 犯罪嫌疑人盧某
 */
public class cMaster {
    /**
     * 建立一個ConcurrentLinkedQueue 用來盛放任務
     * ConcurrentLinkedQueue是一個執行緒安全的無界執行緒安全佇列
     * 對元素的排序遵循先進先出的原則 獲取元素時返回頭部元素 新增元素則為尾部
     */
    private Queue<Object> workersQueue = new ConcurrentLinkedQueue<>();
    /**
     * 建立HashMap<K,V> 來存放Worker物件
     */
    private Map<String, Thread> workers = new HashMap<>();
    /**
     * 建立ConcurrentHashMap來存放Worker計算後的結果集
     */
    private Map<String, Object> resultMap = new ConcurrentHashMap<>();
    /**
     * Master類的有參構造
     * @param worker Worker物件
     * @param workersCount Worker的數量 用來建立對應數量的執行緒
     */
    public cMaster(cWorker worker , int workersCount ) {
        worker.setIntoWorkersQueue(this.workersQueue);
        worker.setResultMap(this.resultMap);
        /*
         * 建立對應數量的執行緒來模擬Worker
         * 至於顯式new Thread() 阿里的程式碼規約什麼的對我這個無業遊民毫無約束 隨它去
         */
        for (int i = 0 ; i < workersCount ; i ++) {
            workers.put(Integer.toString(i) , new Thread(worker , Integer.toString(i)));
        }
    }
    /**
     * 判斷是否所有的子任務都已完成
     * @return 是否完成
     */
    public boolean isComplete() {
        for(Map.Entry<String, Thread> entry : workers.entrySet()) {
            if (entry.getValue().getState() != Thread.State.TERMINATED) {
                return false;
            }
        }
        return true;
    }
    /**
     * 向任務佇列中提交子任務
     */
    public void missionSubmit(Object mission) {
        workersQueue.add(mission);
    }
    /**
     * 返回子任務的結果集
     * @return 子任務執行完畢的結果集
     */
    public Map<String, Object> getResultMap() {
        return resultMap;
    }
    /**
     * 啟動所有的Workers執行緒 開始平行計算
     */
    public void startAllWorkes() {
        for (Map.Entry<String,Thread> threadEntry : workers.entrySet()) {
            threadEntry.getValue().start();
        }
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79

再是Worker:

package com.unsc.Master_Worker;

import java.util.Map;
import java.util.Queue;


/**
 * Created by DELL on 2017/12/25.
 * Master-Worker 中的 Worker類
 * @author 犯罪嫌疑人盧某
 */
public class cWorker implements Runnable{
    /**
     * Worker中的任務佇列
     */
    private Queue<Object> workQueue;
    /**
     * Worker中任務的結果集Map
     */
    private Map<String, Object> resultMap;
    /**
     * 實現Runnable介面重寫的Run方法
     */
    @Override
    public void run() {
        /*
         * 設定輪詢 獲取子任務並且處理
         */
        while (true) {
            Object mission = workQueue.poll();
            if (mission == null) {
                break;
            }
            /*
             * 模擬子任務的處理 並且把處理結果加入結果集
             */
            Object result = executeMission(mission);
            resultMap.put(Integer.toString(mission.hashCode()) , result);
        }
    }
    /**
     * 具體執行任務的業務邏輯 這裡是基本的Worker 具體的邏輯交給子類實現
     * @param mission 任務
     * @return 執行結果
     */
    public Object executeMission(Object mission) {
        return mission;
    }
    /**
     * 設定Worker的任務佇列
     * @param workersQueue 任務佇列
     */
    void setIntoWorkersQueue(Queue<Object> workersQueue) {
        this.workQueue = workersQueue;
    }
    /**
     * 設定Worker的結果集
     * @param resultMap 結果集
     */
    void setResultMap(Map<String, Object> resultMap) {
        this.resultMap = resultMap;
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64

再建立一個Worker的具體的實現類 這裡通過一個TrueWorker類重寫Worker中的executeMission方法執行具體的操作. 
TrueWorker:

package com.unsc.Master_Worker;

import java.util.Map;
import java.util.Set;

/**
 * Created by DELL on 2017/12/25.
 * Master-Worker模式的測試
 * @author 犯罪嫌疑人盧某
 */
public class MasterWorkerTest {
    public static void main(String[] args) {
        /*
         * 設定6個worker和100個子任務 for迴圈裡面的魔法值無視吧
         */
        cMaster master = new cMaster(new TrueWorker(), 4);
        for (int i = 1 ; i < 101 ; i++) {
            master.missionSubmit(i + 0.1);
        }
        /*
         * Master讓所有的Worker開始工作 並且在運作完畢後獲得結果集
         */
        //打一個時間戳 對比測試消耗的時間
        long startTime = System.currentTimeMillis();
        master.startAllWorkes();
        Map<String, Object> resultMap = master.getResultMap();
        long endTime = System.currentTimeMillis();
        double finalScore = 0;
        while (true) {
            Set<String> keySet = resultMap.keySet();
            String key = null;
            for (String s : keySet) {
                key = s;
                break;
            }
            //計算結果變數
            Double score = null;
            if (key != null) {
                //計算結果並且從結果集中刪除
                score = (Double) resultMap.get(key);
                resultMap.remove(key);
            }
            if (score != null) {
                finalScore += score;
            }
            if (master.isComplete() && resultMap.size() == 0) {
                System.out.println("任務執行完畢..");
                break;
            }
        }
        System.out.println("結果的值是 --->>> " + finalScore);
        System.out.println("總計耗時 : " + (endTime - startTime) + " ms");
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56

因為int型別計算太快了 我才用雙精度浮點來運算 排除掉建立執行緒和其他的操作單純就關注任務提交和獲得結果集的耗時 得到的執行結果如圖: 
MW方式

同樣的再用傳統的方法 再來檢測一下效率: 
程式碼如下 :

package com.unsc.Master_Worker;

/**
 * Created by DELL on 2017/12/25.
 * 直接單執行緒在main中計算 1 到 100的平方和
 * @author 犯罪嫌疑人盧某
 */
public class OldMethod {
    public static void main(String[] args) {
        /*
         * 開始的時間戳
         */
        long startTime = System.currentTimeMillis();
        double finalScore = 0;
        for(int i = 0 ; i < 101 ; i++) {
            double num = i + 0.1;
            finalScore += num * num;
        }
        System.out.println("結果是 : ---->>>> " + finalScore);
        long endTime = System.currentTimeMillis();
        System.out.println("耗時總計 : " + (endTime - startTime) + " ms");
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

結果如圖: 
傳統方式

這裡存在一些效能差別 雖然很小 只是1ms的差距 但是在高併發大資料量的情況下 會有比較明顯的效能差距 只是單機上進行測試不是很明顯罷了..

相關推薦

多執行緒並行設計模式之Master-Worker模式

Master-Worker模式的核心思想是在於Master程序和Worker程序各自分擔各自的任務,協同完成資訊處理的模式 使用一個圖例來簡單描述一下:  如圖所示Master-Worker的工作機制就是Master接收到了一個任務,對任務進行拆分,並且分配給各個Worker

多執行緒下的設計模式之Master-Worker模式

該模式可以簡單理解為：首先client將任務交給Master，Master中使用一個併發集合類來承載所有任務，使用一個集合去承載所有的Worker物件，並且有一個併發集合類來承載每一個Worker併發處理任務的結果集；每一個Worker是一個工作執行緒，所以首先要實現Runn

Java多執行緒學習筆記21之單例模式與多執行緒

詳細程式碼見:github程式碼地址   第六章 單例模式與多執行緒 前言: 我之前已經開設了23個設計模式這個專欄，介紹了很多的Java設計模式，其中一些模式對於絕 大多數程式語言設計思想都是類似的，需要了解單例模式的可以去看看。 我們在實際開發中經常用到單例模式，但

多執行緒設計模式：Master-Worker模式

Master-Worker是常用的平行計算模式。它的核心思想是系統由兩類程序協作工作：Master程序和Worker程序。Master負責接收和分配任務，Worker負責處理子任務。當各個Worker子程序處理完成後，會將結果返回給Master，由Master作歸納總結。其好處就是能將一個大任務分解

Java 多執行緒程式設計設計模式之 Thread Pool(執行緒池）

Java 中建立一個執行緒往往意味著JVM 會建立相應依賴於宿主機作業系統的本地執行緒（Native Thread)。 Thread Pool 模式的本質是使用極其有限的資源去處理相對無限的任務。核心思想是使用佇列對待處理的任務進行快取，並複用一定數量的工作者

day24總結_多執行緒和設計模式

1、多執行緒 ①、JDK5以後的針對執行緒的鎖定操作和釋放操作 // 定義鎖物件 private Lock lock = new ReentrantLock(); // 加鎖 lock.lock(); // 釋放鎖 lock.unlock(); ②、死鎖問題的描述和程式碼體現 *死鎖：兩個或

學習彙集地，如果你擅長計算機組成原理，多執行緒，設計模式，jvm，前端或者其他都可以

擅長jvm 多執行緒 設計模式 資料庫 前端 分散式什麼的一起學習共同進步。   目的是大家在自學新的領域的時候有地方可以探討求疑 比如在看到垃圾回收各種收集器中遇到執行緒方面知識的時候 學習設計模式分不清單例和享元的區別，只有書本經驗不知道如何應用到實際開發的時候 學習資料庫系統

多執行緒的設計模式---future模式1

Future模式是多執行緒開發中非常常見的一種設計模式，它的核心思想是非同步呼叫。這類似我們日常生活中的線上購物流程，帶在購物網看著一件商品時可以提交表單，當訂單完成後就可以在家裡等待商品送貨上門。或者說更形象的是我們傳送Ajax請求的時候，頁面是非同步的進行後臺處理，使用者無需等待請求的結果，可以

【多執行緒高併發】多執行緒的設計模式

多執行緒的設計模式 程式碼在Multi_004當中 並行設計模式屬於設計優化的一部分，他是對一些常用的多執行緒結構的總結和抽象，與序列程式相比，並行程式的結構通常更為複雜，因此合理的使用並行模式在多執行緒開發中更具有意義，在這裡主要介紹Futur

將MySQL去重操作優化到極致之三彈連發（二）：多執行緒並行執行

        上一篇已經將單條查重語句調整到最優，但該語句是以單執行緒方式執行。能否利用多處理器，讓去重操作多執行緒並行執行，從而進一步提高速度呢？比如我的實驗環境是4處理器，如果使用4個執行緒同時執行查重sql，理論上應該接近4倍的效能提升。一、資料分片        我

OSG 多執行緒模型 設計思想

A New Processing Model for Multithreaded, Multidisplay Scene Graphs Copyright © 2001 Don Burns (DB - Apr 28, 2004) This article

MFC多執行緒程式設計實踐總結之AfxBeginThread()

在MFC多執行緒程式設計中，執行緒函式呼叫類內成員變數和成員函式的步驟： 1.將執行緒函式在類.h檔案中類內宣告，並用修飾符static修飾； class CtestDlg : public CDialogEx {    public:    &nbs

多執行緒記憶體問題分析之mprotect方法【轉】

轉自：https://blog.csdn.net/agwtpcbox/article/details/53230664 http://www.yebangyu.org/blog/2016/02/01/detectmemoryghostinmultithread/ 多執行緒中的記憶體問題，一直被認為是噩夢般

Java多執行緒學習筆記22之增補拾遺

詳細程式碼見:github程式碼地址   第七章 本章內容: 1. 執行緒組的使用 2. 如何切換執行緒的狀態 3. SimpleDateFormat類與多執行緒的解決方案 4. 如何處理執行緒的異常 執行緒的狀態 執行緒物件在不同的執行時期有不同的狀態

Java多執行緒學習筆記20之定時器Timer

詳細程式碼見:github程式碼地址   本節內容:     定時器Timer的使用及分析 1) 如何實現指定時間執行任務 2) 如何實現按指定週期執行任務   第五章 定時器Timer 定時/計劃

Java多執行緒學習筆記19之Lock的使用

詳細程式碼見:github程式碼地址   本節內容: ReentrantLock 方法講解及ReentrantReadWriteLock讀寫鎖詳解 5）awaitUninterruptibly() 6) awaitUntil(Date deadline

Java多執行緒學習筆記18之Lock的使用

詳細程式碼見:github程式碼地址   本節內容:     公平鎖和非公平鎖以及ReentrantLock常用方法的使用 1) 公平鎖和非公平鎖 2) getHoldCount()、getQueueLength()、getWait

Java多執行緒學習筆記17之Lock的使用

詳細程式碼見:github程式碼地址   本節內容: 1) Lock/ReentrantLock的newCondition()方法及Condition類/await、signal方法文件翻譯 2) Condition實現等待/通知使用(錯誤示例及正

Java多執行緒學習筆記16之Lock的使用

詳細程式碼見:github程式碼地址   本節內容: 1) Lock介面/ReentrantLock類/lock()/unlock()方法文件翻譯     翻譯中包含了此重入鎖與synchronized方法及語句塊之間區別及優缺點

Java多執行緒學習筆記15之執行緒間通訊

詳細程式碼見:github程式碼地址   本節內容: 1）ThreadLocal類的使用     JDK文件及方法翻譯     InheritableThreadLocal的使用   5.