2. 程式人生 > >【多執行緒高併發】多執行緒的設計模式

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

阿新 • • 發佈：2019-01-01

多執行緒的設計模式

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

Future模式

Future模式類似於非同步請求
Future模式Java實現舉例(其實JDK自帶有實現，這裡我先用java來實現)
程式碼在： com.kaishun.height.design014 中
java_future模式

main方法 

public class Main {

    public
 
 static void main(String[] args) throws InterruptedException {

        FutureClient fc = new FutureClient();
        Data data = fc.request("請求引數");
        System.out.println("請求傳送成功!");
        System.out.println("做其他的事情...");

        String result = data.getRequest();
        System.out.println(result);

    }
}

FutureClient類 先返回一個futureData物件，不讓主方法阻塞，然後再讓這個引用去得到耗時的操作的結果

public class FutureClient {

    public Data request(final String queryStr){
        //1 我想要一個代理物件（Data介面的實現類）先返回給傳送請求的客戶端，告訴他請求已經接收到，可以做其他的事情
        final FutureData futureData = new FutureData();
        //2 啟動一個新的執行緒，去載入真實的資料，傳遞給這個代理物件
 

        new Thread(new Runnable() {
            @Override
            public void run() {
                //3 這個新的執行緒可以去慢慢的載入真實物件，然後傳遞給代理物件
                RealData realData = new RealData(queryStr);
                futureData.setRealData(realData);
            }
        }).start();

        return futureData;
    }

}

RealData類，構造方法是一個耗時的類的操作

public class RealData implements Data{

    private String result ;

    public RealData (String queryStr){
        System.out.println("根據'" + queryStr + "'進行查詢，這是一個很耗時的操作..");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("操作完畢，獲取結果");
        result = "'查詢結果'";
    }

    @Override
    public String getRequest() {
        return result;
    }

}

FutureData類 setRealData(RealData realData)和getRequest()加了同步程式碼塊synchronized, 執行緒之間通過wait/notify進行通訊

public class FutureData implements Data{

    private RealData realData ;

    private boolean isReady = false;

    public synchronized void setRealData(RealData realData) {
        //如果已經裝載完畢了，就直接返回
        if(isReady){
            return;
        }
        //如果沒裝載，進行裝載真實物件
        this.realData = realData;
        isReady = true;
        //進行通知
        notify();
    }

    @Override
    public synchronized String getRequest() {
        //如果沒裝載好 程式就一直處於阻塞狀態
        while(!isReady){
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //裝載好直接獲取資料即可
        return this.realData.getRequest();
    }

}

Data介面

public interface Data {

    String getRequest();

}

最終輸出 

請求傳送成功!
做其他的事情...
根據'請求引數'進行查詢，這是一個很耗時的操作..
操作完畢，獲取結果
'查詢結果'

Master-Worker模式

Master-Worker模式是常用的平行計算模式。他的核心思想是系統由兩類程序協作工作：Master程序和Worker程序.Maseter負責接收和分配任務, Worker負責處理子任務。當各個Worker子進行處理完成後，會將結果返回給Master，由Msster做歸納總結，好處是能將一個大任務分解成若干個小任務，並行執行，從而提高系統的吞吐量
java併發Master-Worker模式
舉例com.kaishun.height.design015

Master-Worker模式編寫步驟

Task任務類 

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

}

Master類 

    //1 有一個盛放任務的容器
    private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();

    //2 需要有一個盛放worker的集合
    private HashMap<String, Thread> workers = new HashMap<String, Thread>();

    //3 需要有一個盛放每一個worker執行任務的結果集合
    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 需要有一個執行的方法，啟動所有的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;
    }

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

}

Work類 

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);
            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) {
            e.printStackTrace();
        }
        return output;
    }
}

Main方法 

public class Main {

    public static void main(String[] args) {

        Master master = new Master(new Worker(), 20);

        Random r = new Random();
        // 100 個任務
        for(int i = 1; i <= 100; i++){
            Task t = new Task();
            t.setId(i);
            t.setPrice(r.nextInt(1000));
            master.submit(t);
        }

        master.execute();
        long start = System.currentTimeMillis();

        while(true){
            if(master.isComplete()){
                long end = System.currentTimeMillis() - start;
                int priceResult = master.getResult();
                System.out.println("最終結果：" + priceResult + ", 執行時間：" + end);
                break;
            }
        }

    }
}

最終輸出結果 

最終結果：48098, 執行時間：2500

1.4 生產者-消費者模式

生產者和消費者也是一個非常經典的多執行緒模式，我們在實際中開發應用非常廣泛的思想理念。在生產-消費模式中:通常由兩類執行緒，即若干個生產者和若干個消費者的執行緒。生產者負責提交使用者資料，消費者負責具體處理生產者提交的任務，在生產者和消費者之間通過共享記憶體快取區進行通訊。

高併發生產者-消費者模式

示例：
現在就來模擬一下上面的圖示
main方法

public class Main {

    public static void main(String[] args) throws Exception {
        //記憶體緩衝區
        BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(10);
        //生產者
        Provider p1 = new Provider(queue);
        Provider p2 = new Provider(queue);
        Provider p3 = new Provider(queue);
        //消費者
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        //建立執行緒池執行,這是一個快取的執行緒池，可以建立無窮大的執行緒，沒有任務的時候不建立執行緒。空閒執行緒存活時間為60s（預設值）

        ExecutorService cachePool = Executors.newCachedThreadPool();
        cachePool.execute(p1);
        cachePool.execute(p2);
        cachePool.execute(p3);
        cachePool.execute(c1);
        cachePool.execute(c2);
        cachePool.execute(c3);

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        p1.stop();
        p2.stop();
        p3.stop();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }       
//      cachePool.shutdown(); 
//      cachePool.shutdownNow();

    }

}

Provider 

public class Provider implements Runnable{

    //共享快取區
    private BlockingQueue<Data> queue;
    //多執行緒間是否啟動變數，有強制從主記憶體中重新整理的功能。即時返回執行緒的狀態
    private volatile boolean isRunning = true;
    //id生成器
    private static AtomicInteger count = new AtomicInteger();
    //隨機物件
    private static Random r = new Random(); 

    public Provider(BlockingQueue queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        while(isRunning){
            try {
                //隨機休眠0 - 1000 毫秒 表示獲取資料(產生資料的耗時) 
                Thread.sleep(r.nextInt(1000));
                //獲取的資料進行累計...
                int id = count.incrementAndGet();
                //比如通過一個getData方法獲取了
                Data data = new Data(Integer.toString(id), "資料" + id);
                System.out.println("當前執行緒:" + Thread.currentThread().getName() + ", 獲取了資料，id為:" + id + ", 進行裝載到公共緩衝區中...");
                if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
                    System.out.println("提交緩衝區資料失敗....");
                    //do something... 比如重新提交
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void stop(){
        this.isRunning = false;
    }

}

Consumer 

public class Consumer implements Runnable{

    private BlockingQueue<Data> queue;

    public Consumer(BlockingQueue queue){
        this.queue = queue;
    }

    //隨機物件
    private static Random r = new Random(); 

    @Override
    public void run() {
        while(true){
            try {
                //獲取資料
                Data data = this.queue.take();
                //進行資料處理。休眠0 - 1000毫秒模擬耗時
                Thread.sleep(r.nextInt(1000));
                System.out.println("當前消費執行緒：" + Thread.currentThread().getName() + "， 消費成功，消費資料為id: " + data.getId());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Data類

public final class Data {

    private String id;
    private String name;

    public Data(String id, String name){
        this.id = id;
        this.name = name;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString(){
        return "{id: " + id + ", name: " + name + "}";
    }

}

輸出

當前執行緒:pool-1-thread-2, 獲取了資料，id為:1, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 1
當前執行緒:pool-1-thread-3, 獲取了資料，id為:2, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-2, 獲取了資料，id為:3, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 2
當前執行緒:pool-1-thread-1, 獲取了資料，id為:4, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-2, 獲取了資料，id為:5, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 4
當前消費執行緒：pool-1-thread-4， 消費成功，消費資料為id: 3
當前執行緒:pool-1-thread-1, 獲取了資料，id為:6, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-3, 獲取了資料，id為:7, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 5
當前消費執行緒：pool-1-thread-4， 消費成功，消費資料為id: 7
當前執行緒:pool-1-thread-2, 獲取了資料，id為:8, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-2, 獲取了資料，id為:9, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-1, 獲取了資料，id為:10, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 6
當前執行緒:pool-1-thread-3, 獲取了資料，id為:11, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-1, 獲取了資料，id為:12, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 10
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 8
當前執行緒:pool-1-thread-1, 獲取了資料，id為:13, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-3, 獲取了資料，id為:14, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 12
當前執行緒:pool-1-thread-1, 獲取了資料，id為:15, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-4， 消費成功，消費資料為id: 9
當前執行緒:pool-1-thread-2, 獲取了資料，id為:16, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 13
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 11
當前執行緒:pool-1-thread-3, 獲取了資料，id為:17, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 15
當前消費執行緒：pool-1-thread-4， 消費成功，消費資料為id: 14
當前執行緒:pool-1-thread-2, 獲取了資料，id為:18, 進行裝載到公共緩衝區中...
當前執行緒:pool-1-thread-1, 獲取了資料，id為:19, 進行裝載到公共緩衝區中...
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 16
當前消費執行緒：pool-1-thread-6， 消費成功，消費資料為id: 17
當前消費執行緒：pool-1-thread-5， 消費成功，消費資料為id: 19
當前消費執行緒：pool-1-thread-4， 消費成功，消費資料為id: 18

特別感謝網際網路架構師白鶴翔老師，本文大多出自他的講解。
筆者主要是記錄筆記，正所謂好記性不如爛筆頭，爛筆頭不如雲筆記

相關推薦

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

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

【多執行緒高併發】執行緒安全

關鍵字：執行緒安全，synchronized，多個執行緒多個鎖，物件鎖的同步和非同步，髒讀，synchronized鎖重入，synchronized程式碼塊，volatile關鍵字 執行緒安全 當多個執行緒訪問某一個類（物件或者方法）時，這個累始終都能表現出

【多執行緒高併發】 同步容器和佇列

關鍵字: 同步容器，佇列，ConcurrentMap, Copy-On-Write容器,併發Queue, ConcurrentLinkedQueue, BlockQueue介面, ArrayBlockingQueue , LinkedBlockingQue

網際網路高併發架構的8種設計模式演化

1、單庫單應用模式這種是最簡單的模式，即一個數據一個應用伺服器，一般在產品釋出初期使用會比較方便，單日30萬到50萬PV以下一般沒有問題。2、內容分發模式在主機中使用了靜態檔案快取之後，還可以使用CDN的方式把靜態檔案分發到離使用者最近的節點上以達到快速響應的目的，一般在百萬

【Django】Uwsgi+Nginx+Django2.0+Python3.7實現高併發，多執行緒，高效能

一、系統以及環境 伺服器系統：Ubuntu 16.04 專案環境：python 3.7 框架：Django2.0 伺服器環境：Uwsgi、Nginx 效能監測工具：Uwsgitop 使用背景：因為Dj

【本人禿頂程式設計師】你分得清分散式、高併發與多執行緒嗎？

←←←←←←←←←←←← 快，點關注！ 當提起這三個詞的時候，是不是很多人都認為分散式=高併發=多執行緒？ 當面試官問到高併發系統可以採用哪些手段來解決，或者被問到分散式系統如何解決一致性的問題，是不是一臉懵逼？ 確實，在一開始接觸的時候，不少人都會將三者混淆，誤以為所謂的分散式

【java併發】多個執行緒間共享資料

先看一個多執行緒間共享資料的問題： 設計四個執行緒，其中兩個執行緒每次對data增加1，另外兩個執行緒每次對data減少1。 　　從問題來看，很明顯涉及到了執行緒間通資料的共享，四個執行

【併發】多執行緒程式設計中條件變數和虛假喚醒的討論

轉自：http://blog.csdn.net/puncha/article/details/8493862 From: http://siwind.iteye.com/blog/1469216 From:http://en.wikipedia.org/wiki/S

SimpleDateFormat時間格式化高併發、多執行緒時出現問題

SimpleDateFormat是是 Java 中一個非常常用的類，該類用來對日期字串進行解析和格式化輸出，但是DateFormat 和 SimpleDateFormat 類不都是執行緒安全的，在生產環境的多執行緒或高併發情況使用 format() 和 parse() 方法，會出現很多問題：

SpringBoot實戰實現分散式鎖一之重現多執行緒高併發場景

實戰前言：上篇博文我總體介紹了我這套視訊課程：“SpringBoot實戰實現分散式鎖” 總體涉及的內容，從本篇文章開始，我將開始介紹其中涉及到的相關知識要點，感興趣的小夥伴可以關注關注學習學習！！工欲善其事，必先利其器，介紹分散式鎖使用的前因後果之前，得先想辦法說清楚為啥需要分散式鎖以及

高併發與多執行緒的關係、區別、高併發的技術方案

高併發與多執行緒的關係、區別、高併發的技術方案 http://youzhixueyuan.com/high-concurrency-and-multithreading-high-concurrency-technical-solutions.html 什麼是高併發？ 高併發（High

Java高併發程式設計——多執行緒

執行緒簡介 程序 正在執行的程式。 程序是系統進行資源分配和呼叫的獨立單位。每一個程序都有他自己的記憶體空間和系統資源。 執行緒 在同一個程序內又可以執行多個任務。而這每一個任務就可以看成是一個執行緒。

Java 多執行緒高併發 1 — 幾個重要的概念

原子性： 和物理上的原子貌似沒多大關係，對於一系列的操作只能單個執行緒做處理，不能有其他執行緒干擾，例如多個人想往同一個杯子倒茶，但只有一個茶壺，只有等這個茶壺讓出來才能繼續讓其他人倒茶 有序性：

Java 多執行緒高併發 2 — CAS 無鎖

在 Java 併發程式設計裡面，最可愛的就是無鎖了，非常巧妙，精彩絕倫 額。O__O "… 那麼什麼是無鎖？ 顧名思義，在併發情況下采用無鎖的方式實現物件操作的原子性，保證資料一致性、安全性、正確性

Java 多執行緒高併發 3.3 — Semaphore 共享鎖使用

Semaphore 是共享鎖的一種，字面上意思就是訊號量鎖 顧名思義，一個可以共享的鎖，可以讓多個執行緒共享同一把鎖，例如同一條馬路可以讓 4 臺車同時並行跑，相當於可以讓 4 個執行緒共享一把鎖，臨

Java 多執行緒高併發 3.5 — ReadWriteLock 讀寫鎖使用

ReadWriteLock 又稱為共享鎖，的讀寫分離鎖，內部分為讀鎖和寫鎖 關於 A 和 B 執行緒，讀和寫的關係 A 讀 — B 讀：不互斥、不阻塞 A 讀 — B 寫：互斥，互相阻塞 A 寫 — B 寫：互斥，互相阻塞 用法：非常簡單，和 Reentra

【本人禿頂程式設計師】多執行緒：為什麼在while迴圈中加入System.out.println，執行緒可以停止

在論壇看到這樣一個程式碼： public class StopThread { private static boolean stopRequested; public static void main(String[] args) throws InterruptedE

java多執行緒高併發知識總結

1.      計算機系統 使用快取記憶體來作為記憶體與處理器之間的緩衝，將運算需要用到的資料複製到快取中，讓計算能快速進行；當運算結束後再從快取同步回記憶體之中，這樣處理器就無需等待緩慢的記憶體讀寫了。 快取一致性：多處理器系統中，因為共享同一主記憶體，

多執行緒高併發訪問--------synchronized和ThreadLocal的區別

synchronized關鍵字主要解決多執行緒共享資料同步問題。ThreadLocal使用場合主要解決多執行緒中資料因併發產生不一致問題。 ThreadLocal和Synchonized都用於解決多執行緒併發訪問。但是ThreadLocal與synchronized有本質的區別: &nb

專案中怎麼控制多執行緒高併發訪問

synchronized關鍵字主要解決多執行緒共享資料同步問題。 ThreadLocal使用場合主要解決多執行緒中資料因併發產生不一致問題。 ThreadLocal和Synchonized都用於解決多執行緒併發訪問。但是ThreadLocal與synchronized有