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

来源: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
  • WPF 已知問題 監聽 WMI 事件導致觸摸失效
    示例項目結構 在 Visual Studio 中創建一個 WinForms 應用程式後，項目結構如下所示： MyWinFormsApp/ │ ├───Properties/ │ └───Settings.settings │ ├───bin/ │ ├───Debug/ │ └───Release/ ...
  • [STAThread]特性
    [STAThread] 特性用於需要與 COM 組件交互的應用程式，尤其是依賴單線程模型（如 Windows Forms 應用程式）的組件。在 STA 模式下，線程擁有自己的消息迴圈，這對於處理用戶界面和某些 COM 組件是必要的。 [STAThread] static void Main(stri ...
  • UWP WinUI3 傳入 AddHandler 的 RoutedEventHandler 類型與事件所需不匹配將拋出參數異常
    在WinForm中使用全局異常捕獲處理 在WinForm應用程式中，全局異常捕獲是確保程式穩定性的關鍵。通過在Program類的Main方法中設置全局異常處理，可以有效地捕獲並處理未預見的異常，從而避免程式崩潰。 註冊全局異常事件 [STAThread] static void Main() { / ...
  • .NET 多版本 WinForm 開源控制項庫 SunnyUI
    前言 給大家推薦一款開源的 Winform 控制項庫，可以幫助我們開發更加美觀、漂亮的 WinForm 界面。 項目介紹 SunnyUI.NET 是一個基於 .NET Framework 4.0+、.NET 6、.NET 7 和 .NET 8 的 WinForm 開源控制項庫，同時也提供了工具類庫、擴展 ...
  • .net core8 使用Swagger（附當前源碼）
    說明 該文章是屬於OverallAuth2.0系列文章，每周更新一篇該系列文章（從0到1完成系統開發）。 該系統文章，我會儘量說的非常詳細，做到不管新手、老手都能看懂。 說明：OverallAuth2.0 是一個簡單、易懂、功能強大的許可權+可視化流程管理系統。 有興趣的朋友，請關註我吧(*^▽^*) ...
  • WPF 什麼時候 VisualTreeHelper.GetDescendantBounds 將返回無窮大
    一、下載安裝 1.下載git 必須先下載並安裝git，再TortoiseGit下載安裝 git安裝參考教程：https://blog.csdn.net/mukes/article/details/115693833 2.TortoiseGit下載與安裝 TortoiseGit，Git客戶端，32/6 ...
  • C# 開源教程帶你輕鬆掌握數據結構與演算法
    前言 在項目開發過程中，理解數據結構和演算法如同掌握蓋房子的秘訣。演算法不僅能幫助我們編寫高效、優質的代碼，還能解決項目中遇到的各種難題。 給大家推薦一個支持C#的開源免費、新手友好的數據結構與演算法入門教程：Hello演算法。 項目介紹 《Hello Algo》是一本開源免費、新手友好的數據結構與演算法入門 ...
  • 第22篇 生成proto文件bat腳本
    1.生成單個Proto.bat內容 @rem Copyright 2016, Google Inc. @rem All rights reserved. @rem @rem Redistribution and use in source and binary forms, with or with ...
  • 記一次 .NET某上位機視覺程式 卡死分析
    一：背景 1. 講故事 前段時間有位朋友找到我，說他的窗體程式在客戶這邊出現了卡死，讓我幫忙看下怎麼回事？dump也生成了，既然有dump了那就上 windbg 分析吧。 二：WinDbg 分析 1. 為什麼會卡死 窗體程式的卡死，入口門檻很低，後續往下分析就不一定了，不管怎麼說先用 !clrsta ...
  • 醫學DICOM文件與PACS系統進行數據傳輸
    前言 人工智慧時代，人臉識別技術已成為安全驗證、身份識別和用戶交互的關鍵工具。 給大家推薦一款.NET 開源提供了強大的人臉識別 API，工具不僅易於集成，還具備高效處理能力。 本文將介紹一款如何利用這些API，為我們的項目添加智能識別的亮點。 項目介紹 GitHub 上擁有 1.2k 星標的 C# ...
所有分類