多線程系列(十二) -生產者和消費者模型

来源:https://www.cnblogs.com/dxflqm/p/18051189
-Advertisement-
Play Games

一、簡介 在 Java 多線程編程中，還有一個非常重要的設計模式，它就是：生產者和消費者模型。 這種模型可以充分發揮 cpu 的多線程特性，通過一些平衡手段能有效的提升系統整體處理數據的速度，減輕系統負載，提高程式的效率和穩定性，同時實現模塊之間的解耦。 那什麼是生產者和消費者模型呢？ 簡單的說，生 ...


一、簡介

在 Java 多線程編程中，還有一個非常重要的設計模式，它就是：生產者和消費者模型。

這種模型可以充分發揮 cpu 的多線程特性，通過一些平衡手段能有效的提升系統整體處理數據的速度，減輕系統負載，提高程式的效率和穩定性，同時實現模塊之間的解耦。

那什麼是生產者和消費者模型呢？

簡單的說，生產者和消費者之間不直接進行交互，而是通過一個緩衝區來進行交互，生產者負責生成數據，然後存入緩衝區；消費者則負責處理數據，從緩衝區獲取。

大致流程圖如下：

對於最簡單的生產者和消費者模型，總結下來，大概有以下幾個特點：

  • 緩衝區為空的時候，消費者不能消費，會進入休眠狀態，直到有新數據進入緩衝區，再次被喚醒
  • 緩衝區填滿的時候，生產者不能生產，也會進入休眠狀態，直到緩衝區有空間，再次被喚醒

生產者和消費者模型作為一個非常重要的設計模型，它的優點在於：

  • 解耦：生產者和消費者之間不直接進行交互，即使生產者和消費者的代碼發生變化，也不會對對方產生影響
  • 消峰：例如在某項工作中，假如 A 操作生產數據的速度很快，B 操作處理速度很慢，那麼 A 操作就必須等待 B 操作完成才能結束，反之亦然。如果將 A 操作和B 操作進行解耦，中間插入一個緩衝區，這樣 A 操作將生產的數據存入緩衝區，就接受了；B 操作從緩衝區獲取數據併進行處理，平衡好 A 操作和 B 操作之間的緩衝區，可以顯著提升系統的數據處理能力

生產者和消費者模型的應用場景非常多，例如 Java 的線程池任務執行框架、消息中間件 rabbitMQ 等，因此掌握生產者和消費者模型，對於開發者至關重要。

下麵我們通過幾個案例，一起來瞭解一下生產者和消費者設計模型的實踐思路。

二、代碼實踐

2.1、利用 wait / notify 方法實現思路

生產者和消費者模型，最簡單的一種技術實踐方案就是基於線程的 wait() / notify() 方法，也就是通知和喚醒機制，可以將兩個操作實現解耦，具體代碼實踐如下。

/**
 * 緩衝區容器類
 */
public class Container {

    /**
     * 緩衝區最大容量
     */
    private int capacity = 3;

    /**
     * 緩衝區
     */
    private LinkedList<Integer> list = new LinkedList<Integer>();


    /**
     * 添加數據到緩衝區
     * @param value
     */
    public synchronized void add(Integer value) {
        if(list.size() >= capacity){
            System.out.println("生產者："+ Thread.currentThread().getName()+"，緩衝區已滿,生產者進入waiting...");
            try {
                // 進入等待狀態
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("生產者："+ Thread.currentThread().getName()+"，add：" + value);
        list.add(value);

        //喚醒其他所有處於wait()的線程，包括消費者和生產者
        notifyAll();
    }


    /**
     * 從緩衝區獲取數據
     */
    public synchronized void get() {
        if(list.size() == 0){
            System.out.println("消費者："+ Thread.currentThread().getName()+"，緩衝區為空,消費者進入waiting...");
            try {
                // 進入等待狀態
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 從頭部獲取數據，並移除元素
        Integer val = list.removeFirst();
        System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);

        //喚醒其他所有處於wait()的線程，包括消費者和生產者
        notifyAll();
    }
}

/**
 * 生產者類
 */
public class Producer extends Thread{

    private Container container;

    public Producer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.add(i);
        }
    }
}

/**
 * 消費者類
 */
public class Consumer extends Thread{

    private Container container;

    public Consumer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.get();
        }
    }
}

/**
 * 測試類
 */
public class MyThreadTest {

    public static void main(String[] args) {
        Container container = new Container();
        Producer producer = new Producer(container);
        Consumer consumer = new Consumer(container);

        producer.start();
        consumer.start();
    }
}

運行結果如下：

生產者：Thread-0，add：0
生產者：Thread-0，add：1
生產者：Thread-0，add：2
生產者：Thread-0，緩衝區已滿,生產者進入waiting...
消費者：Thread-1，value：0
消費者：Thread-1，value：1
消費者：Thread-1，value：2
消費者：Thread-1，緩衝區為空,消費者進入waiting...
生產者：Thread-0，add：3
生產者：Thread-0，add：4
生產者：Thread-0，add：5
消費者：Thread-1，value：3
消費者：Thread-1，value：4
消費者：Thread-1，value：5

從日誌上可以很清晰的看到，生產者線程生產一批數據之後，當緩衝區已經滿了，會進入等待狀態，此時會通知消費者線程；消費者線程處理完數據之後，當緩衝區沒有數據時，也會進入等待狀態，再次通知生產者線程。

2.2、利用 await / signal 方法實現思路

除此之外，我們還可以利用ReentrantLock和Condition類中的 await() / signal() 方法實現生產者和消費者模型。

緩衝區容器類，具體代碼實踐如下。

/**
 * 緩衝區容器類
 */
public class Container {

    private Lock lock = new ReentrantLock();

    private Condition condition = lock.newCondition();

    private int capacity = 3;

    private LinkedList<Integer> list = new LinkedList<Integer>();


    /**
     * 添加數據到緩衝區
     * @param value
     */
    public void add(Integer value) {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() >= capacity){
                System.out.println("生產者："+ Thread.currentThread().getName()+"，緩衝區已滿,生產者進入waiting...");
                // 進入等待狀態
                condition.await();
            }
            System.out.println("生產者："+ Thread.currentThread().getName()+"，add：" + value);
            list.add(value);

            //喚醒其他所有處於wait()的線程，包括消費者和生產者
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }


    /**
     * 從緩衝區獲取數據
     */
    public void get() {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() == 0){
                System.out.println("消費者："+ Thread.currentThread().getName()+"，緩衝區為空,消費者進入waiting...");
                // 進入等待狀態
                condition.await();
            }
            // 從頭部獲取數據，並移除元素
            Integer val = list.removeFirst();
            System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);

            //喚醒其他所有處於wait()的線程，包括消費者和生產者
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }
}

生產者、消費者、測試類代碼，跟上文一致，運行結果和上文介紹的也是一樣。

2.3、多生產者和消費者的實現思路

上面介紹的都是一個生產者線程和一個消費者線程，模型比較簡單。實際上，在業務開發中，經常會出現多個生產者線程和多個消費者線程，按照以上的實現思路，會出現什麼情況呢？

有可能會出現程式假死現象！下麵我們來分析一下案例，假如有兩個生產者線程 a1、a2，兩個消費者線程 b1、b2，執行過程如下：

  • 1.生產者線程 a1 執行生產數據的操作，發現緩衝區數據已經填滿了，然後進入等待階段，同時向外發起通知，喚醒其它線程
  • 2.因為線程喚醒具有隨機性，本應該喚醒消費者線程 b1，結果可能生產者線程 a2 被喚醒，檢查緩衝區數據已經填滿了，又進入等待階段，緊接向外發起通知，消費者線程得不到被執行的機會
  • 3.消費者線程 b1、b2，也有可能會出現這個現象，本應該喚醒生產者線程，結果喚醒了消費者線程

遇到這種情況，應該如何解決呢？

因為ReentrantLock和Condition的結合，編程具有高度靈活性，我們可以採用這種組合解決多生產者和多消費者中的假死問題。

具體實現邏輯如下：

/**
 * 緩衝區容器類
 */
public class ContainerDemo {

    private Lock lock = new ReentrantLock();
    private Condition producerCondition = lock.newCondition();
    private Condition consumerCondition = lock.newCondition();

    private int capacity = 3;
    private LinkedList<Integer> list = new LinkedList<Integer>();


    /**
     * 添加數據到緩衝區
     * @param value
     */
    public void add(Integer value) {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() >= capacity){
                System.out.println("生產者："+ Thread.currentThread().getName()+"，緩衝區已滿,生產者進入waiting...");
                // 生產者進入等待狀態
                producerCondition.await();
            }
            System.out.println("生產者："+ Thread.currentThread().getName()+"，add：" + value);
            list.add(value);

            // 喚醒所有消費者處於wait()的線程
            consumerCondition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }


    /**
     * 從緩衝區獲取數據
     */
    public void get() {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() == 0){
                System.out.println("消費者："+ Thread.currentThread().getName()+"，緩衝區為空,消費者進入waiting...");
                // 消費者進入等待狀態
                consumerCondition.await();
            }
            // 從頭部獲取數據，並移除元素
            Integer val = list.removeFirst();
            System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);

            // 喚醒所有生產者處於wait()的線程
            producerCondition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }
}

/**
 * 生產者
 */
public class Producer extends Thread{

    private ContainerDemo container;

    private Integer value;

    public Producer(ContainerDemo container, Integer value) {
        this.container = container;
        this.value = value;
    }

    @Override
    public void run() {
        container.add(value);
    }
}

/**
 * 消費者
 */
public class Consumer extends Thread{

    private ContainerDemo container;

    public Consumer(ContainerDemo container) {
        this.container = container;
    }

    @Override
    public void run() {
        container.get();
    }
}


/**
 * 測試類
 */
public class MyThreadTest {

    public static void main(String[] args) {
        ContainerDemo container = new ContainerDemo();

        List<Thread> threadList = new ArrayList<>();
        // 初始化6個生產者線程
        for (int i = 0; i < 6; i++) {
            threadList.add(new Producer(container, i));
        }
        // 初始化6個消費者線程
        for (int i = 0; i < 6; i++) {
            threadList.add(new Consumer(container));
        }

        // 啟動線程
        for (Thread thread : threadList) {
            thread.start();
        }
    }
}

運行結果如下：

生產者：Thread-0，add：0
生產者：Thread-1，add：1
生產者：Thread-2，add：2
生產者：Thread-3，緩衝區已滿,生產者進入waiting...
生產者：Thread-4，緩衝區已滿,生產者進入waiting...
生產者：Thread-5，緩衝區已滿,生產者進入waiting...
消費者：Thread-6，value：0
消費者：Thread-7，value：1
生產者：Thread-3，add：3
生產者：Thread-4，add：4
生產者：Thread-5，add：5
消費者：Thread-8，value：2
消費者：Thread-9，value：3
消費者：Thread-10，value：4
消費者：Thread-11，value：5

通過ReentrantLock定義兩個Condition，一個表示生產者的Condition，一個表示消費者的Condition，喚醒的時候調用對應的signalAll()方法就可以解決假死現象。

三、小結

最後我們來總結一下，對於生產者和消費者模型，通過合理的編程實現，可以充分充分發揮 cpu 多線程的特性，顯著的提升系統處理數據的效率。

對於生產者和消費者模型中的假死現象，可以使用ReentrantLock定義兩個Condition，進行交叉喚醒，以解決假死問題。

四、參考

1、https://www.cnblogs.com/xrq730/p/4855663.html


作者：程式員志哥
出處：pzblog.cn
資源：微信搜【程式員志哥】關註我，回覆 【技術資料】有我準備的一線程式必備電腦書籍、大廠面試資料和免費電子書。 希望可以幫助大家提升技術和能力。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 003-Java程式流程式控制制
    3. Java程式流程式控制制(重點) 程式的三種控制結構 3.1 分支結構 if, switch 3.1.1 if if 分支 根據條件（真或假）來決定執行某段代碼。 if分支應用場景 if 第一種形式 執行流程： 首先判斷條件表達式的結果，如果為true執行語句體，為false就不執行語句體。 if ...
  • MySQL-18 MySQL8其他新特性
    C-18.MySQL8其他新特性 1.MySQL8新特性概述 MySQL從5.7版本直接跳躍發佈了8.0版本，可見是一個令人興奮的里程碑的版本。MySQL 8版本在功能上，做了顯著的改進與增強，開發者對MySQL的源代碼進行了重構，最突出的一點是對MySQL Optimizer優化器進行了改進。不僅 ...
  • C++ mySQL資料庫連接池(windows平臺)
    C++ MySQL資料庫連接池 新手學了C++多線程，看了些資料練手寫了C++資料庫連接池小項目，自己的源碼地址 關鍵技術點 MySQL資料庫編程、單例模式、queue隊列容器、C++11多線程編程、線程互斥、線程同步通信和 unique_lock、基於CAS的原子整形、智能指針shared_ptr ...
  • 別再低效篩選數據了！試試pandas query函數
    數據過濾在數據分析過程中具有極其重要的地位，因為在真實世界的數據集中，往往存在重覆、缺失或異常的數據。pandas提供的數據過濾功能可以幫助我們輕鬆地識別和處理這些問題數據，從而確保數據的質量和準確性。 今天介紹的query函數，為我們提供了強大靈活的數據過濾方式，有助於從複雜的數據集中提取有價值的 ...
  • C語言初學習——易錯點合集（持續更新中）
    目錄數組（Array）一、數組概念二、如何聲明一個數組三、如何為數組初始化1、數組本身初始化：2、數組的元素初始化2.1 一維數組2.2多維數組四、如何表示數組的各個概念五、數組記憶體和分配空間六、數組相關演算法七、十大內部排序演算法八、數組的工具類：Arrays九、數組的異常 數組（Array） 一、數 ...
  • C++系列：const關鍵字
    前言 在學習C++時，const關鍵字的知識點分散在書的各個章節。當我們嘗試在編程時使用const時，總會感覺有一些細節被遺忘，因而不能得心應手地使用const關鍵字。因此，本篇文章嘗試著對const關鍵字的做一些總結。參考書籍《C++ Primer Plus》 const總結 這裡是我做的關於co ...
  • 解密Spring中的Bean實例化：推斷構造方法（上）
    在Spring中，實例化Bean對象涉及構造方法的調用。通過分析源碼，我們瞭解到實例化的步驟和推斷構造方法的過程。當一個類只有一個構造方法時，Spring會根據具體情況決定是否使用該構造方法。如果一個類存在多個構造方法，就需要根據具體情況具體分析。 ...
  • C++類開發第七篇（詳細說說多態和編譯原理）
    polymorphism 靜態聯編和動態聯編 多態性(polymorphism)提供介面與具體實現之間的另一層隔離，從而將”what”和”how”分離開來。多態性改善了代碼的可讀性和組織性，同時也使創建的程式具有可擴展性，項目不僅在最初創建時期可以擴展，而且當項目在需要有新的功能時也能擴展。 c++ ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類