2. 程式人生 > >C++11 多執行緒下生產者消費者模型詳解

C++11 多執行緒下生產者消費者模型詳解

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



前面八章介紹了 C++11 併發程式設計的基礎(抱歉哈，第五章-第八章還在草稿中)，本文將綜合運用 C++11 中的新的基礎設施(主要是多執行緒、鎖、條件變數)來闡述一個經典問題——生產者消費者模型，並給出完整的解決方案。

生產者消費者問題是多執行緒併發中一個非常經典的問題，相信學過作業系統課程的同學都清楚這個問題的根源。本文將就四種情況分析並介紹生產者和消費者問題，它們分別是：單生產者-單消費者模型，單生產者-多消費者模型，多生產者-單消費者模型，多生產者-多消費者模型，我會給出四種情況下的 C++11 併發解決方案，如果文中出現了錯誤或者你對程式碼有異議，歡迎交流 ;-)。

單生產者-單消費者模型

顧名思義，單生產者-單消費者模型中只有一個生產者和一個消費者，生產者不停地往產品庫中放入產品，消費者則從產品庫中取走產品，產品庫容積有限制，只能容納一定數目的產品，如果生產者生產產品的速度過快，則需要等待消費者取走產品之後，產品庫不為空才能繼續往產品庫中放置新的產品，相反，如果消費者取走產品的速度過快，則可能面臨產品庫中沒有產品可使用的情況，此時需要等待生產者放入一個產品後，消費者才能繼續工作。C++11實現單生產者單消費者模型的程式碼如下：

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int kItemRepositorySize  = 10; // Item buffer size.
static const int kItemsToProduce  = 1000;   // How many items we plan to produce.

struct ItemRepository {
    int item_buffer[kItemRepositorySize]; // 產品緩衝區, 配合 read_position 和 write_position 模型環形佇列.
    size_t read_position; // 消費者讀取產品位置.
    size_t write_position; // 生產者寫入產品位置.
    std::mutex mtx; // 互斥量,保護產品緩衝區
    std::condition_variable repo_not_full; // 條件變數, 指示產品緩衝區不為滿.
    std::condition_variable repo_not_empty; // 條件變數, 指示產品緩衝區不為空.
} gItemRepository; // 產品庫全域性變數, 生產者和消費者操作該變數.

typedef struct ItemRepository ItemRepository;


void ProduceItem(ItemRepository *ir, int item)
{
    std::unique_lock<std::mutex> lock(ir->mtx);
    while(((ir->write_position + 1) % kItemRepositorySize)
        == ir->read_position) { // item buffer is full, just wait here.
        std::cout << "Producer is waiting for an empty slot...\n";
        (ir->repo_not_full).wait(lock); // 生產者等待"產品庫緩衝區不為滿"這一條件發生.
    }

    (ir->item_buffer)[ir->write_position] = item; // 寫入產品.
    (ir->write_position)++; // 寫入位置後移.

    if (ir->write_position == kItemRepositorySize) // 寫入位置若是在佇列最後則重新設定為初始位置.
        ir->write_position = 0;

    (ir->repo_not_empty).notify_all(); // 通知消費者產品庫不為空.
    lock.unlock(); // 解鎖.
}

int ConsumeItem(ItemRepository *ir)
{
    int data;
    std::unique_lock<std::mutex> lock(ir->mtx);
    // item buffer is empty, just wait here.
    while(ir->write_position == ir->read_position) {
        std::cout << "Consumer is waiting for items...\n";
        (ir->repo_not_empty).wait(lock); // 消費者等待"產品庫緩衝區不為空"這一條件發生.
    }

    data = (ir->item_buffer)[ir->read_position]; // 讀取某一產品
    (ir->read_position)++; // 讀取位置後移

    if (ir->read_position >= kItemRepositorySize) // 讀取位置若移到最後，則重新置位.
        ir->read_position = 0;

    (ir->repo_not_full).notify_all(); // 通知消費者產品庫不為滿.
    lock.unlock(); // 解鎖.

    return data; // 返回產品.
}


void ProducerTask() // 生產者任務
{
    for (int i = 1; i <= kItemsToProduce; ++i) {
        // sleep(1);
        std::cout << "Produce the " << i << "^th item..." << std::endl;
        ProduceItem(&gItemRepository, i); // 迴圈生產 kItemsToProduce 個產品.
    }
}

void ConsumerTask() // 消費者任務
{
    static int cnt = 0;
    while(1) {
        sleep(1);
        int item = ConsumeItem(&gItemRepository); // 消費一個產品.
        std::cout << "Consume the " << item << "^th item" << std::endl;
        if (++cnt == kItemsToProduce) break; // 如果產品消費個數為 kItemsToProduce, 則退出.
    }
}

void InitItemRepository(ItemRepository *ir)
{
    ir->write_position = 0; // 初始化產品寫入位置.
    ir->read_position = 0; // 初始化產品讀取位置.
}

int main()
{
    InitItemRepository(&gItemRepository);
    std::thread producer(ProducerTask); // 建立生產者執行緒.
    std::thread consumer(ConsumerTask); // 建立消費之執行緒.
    producer.join();
    consumer.join();
}

 單生產者-多消費者模型

與單生產者和單消費者模型不同的是，單生產者-多消費者模型中可以允許多個消費者同時從產品庫中取走產品。所以除了保護產品庫在多個讀寫執行緒下互斥之外，還需要維護消費者取走產品的計數器，程式碼如下:

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int kItemRepositorySize  = 4; // Item buffer size.
static const int kItemsToProduce  = 10;   // How many items we plan to produce.

struct ItemRepository {
    int item_buffer[kItemRepositorySize];
    size_t read_position;
    size_t write_position;
    size_t item_counter;
    std::mutex mtx;
    std::mutex item_counter_mtx;
    std::condition_variable repo_not_full;
    std::condition_variable repo_not_empty;
} gItemRepository;

typedef struct ItemRepository ItemRepository;


void ProduceItem(ItemRepository *ir, int item)
{
    std::unique_lock<std::mutex> lock(ir->mtx);
    while(((ir->write_position + 1) % kItemRepositorySize)
        == ir->read_position) { // item buffer is full, just wait here.
        std::cout << "Producer is waiting for an empty slot...\n";
        (ir->repo_not_full).wait(lock);
    }

    (ir->item_buffer)[ir->write_position] = item;
    (ir->write_position)++;

    if (ir->write_position == kItemRepositorySize)
        ir->write_position = 0;

    (ir->repo_not_empty).notify_all();
    lock.unlock();
}

int ConsumeItem(ItemRepository *ir)
{
    int data;
    std::unique_lock<std::mutex> lock(ir->mtx);
    // item buffer is empty, just wait here.
    while(ir->write_position == ir->read_position) {
        std::cout << "Consumer is waiting for items...\n";
        (ir->repo_not_empty).wait(lock);
    }

    data = (ir->item_buffer)[ir->read_position];
    (ir->read_position)++;

    if (ir->read_position >= kItemRepositorySize)
        ir->read_position = 0;

    (ir->repo_not_full).notify_all();
    lock.unlock();

    return data;
}


void ProducerTask()
{
    for (int i = 1; i <= kItemsToProduce; ++i) {
        // sleep(1);
        std::cout << "Producer thread " << std::this_thread::get_id()
            << " producing the " << i << "^th item..." << std::endl;
        ProduceItem(&gItemRepository, i);
    }
    std::cout << "Producer thread " << std::this_thread::get_id()
                << " is exiting..." << std::endl;
}

void ConsumerTask()
{
    bool ready_to_exit = false;
    while(1) {
        sleep(1);
        std::unique_lock<std::mutex> lock(gItemRepository.item_counter_mtx);
        if (gItemRepository.item_counter < kItemsToProduce) {
            int item = ConsumeItem(&gItemRepository);
            ++(gItemRepository.item_counter);
            std::cout << "Consumer thread " << std::this_thread::get_id()
                << " is consuming the " << item << "^th item" << std::endl;
        } else ready_to_exit = true;
        lock.unlock();
        if (ready_to_exit == true) break;
    }
    std::cout << "Consumer thread " << std::this_thread::get_id()
                << " is exiting..." << std::endl;
}

void InitItemRepository(ItemRepository *ir)
{
    ir->write_position = 0;
    ir->read_position = 0;
    ir->item_counter = 0;
}

int main()
{
    InitItemRepository(&gItemRepository);
    std::thread producer(ProducerTask);
    std::thread consumer1(ConsumerTask);
    std::thread consumer2(ConsumerTask);
    std::thread consumer3(ConsumerTask);
    std::thread consumer4(ConsumerTask);

    producer.join();
    consumer1.join();
    consumer2.join();
    consumer3.join();
    consumer4.join();
}

 多生產者-單消費者模型

與單生產者和單消費者模型不同的是，多生產者-單消費者模型中可以允許多個生產者同時向產品庫中放入產品。所以除了保護產品庫在多個讀寫執行緒下互斥之外，還需要維護生產者放入產品的計數器，程式碼如下:

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int kItemRepositorySize  = 4; // Item buffer size.
static const int kItemsToProduce  = 10;   // How many items we plan to produce.

struct ItemRepository {
    int item_buffer[kItemRepositorySize];
    size_t read_position;
    size_t write_position;
    size_t item_counter;
    std::mutex mtx;
    std::mutex item_counter_mtx;
    std::condition_variable repo_not_full;
    std::condition_variable repo_not_empty;
} gItemRepository;

typedef struct ItemRepository ItemRepository;


void ProduceItem(ItemRepository *ir, int item)
{
    std::unique_lock<std::mutex> lock(ir->mtx);
    while(((ir->write_position + 1) % kItemRepositorySize)
        == ir->read_position) { // item buffer is full, just wait here.
        std::cout << "Producer is waiting for an empty slot...\n";
        (ir->repo_not_full).wait(lock);
    }

    (ir->item_buffer)[ir->write_position] = item;
    (ir->write_position)++;

    if (ir->write_position == kItemRepositorySize)
        ir->write_position = 0;

    (ir->repo_not_empty).notify_all();
    lock.unlock();
}

int ConsumeItem(ItemRepository *ir)
{
    int data;
    std::unique_lock<std::mutex> lock(ir->mtx);
    // item buffer is empty, just wait here.
    while(ir->write_position == ir->read_position) {
        std::cout << "Consumer is waiting for items...\n";
        (ir->repo_not_empty).wait(lock);
    }

    data = (ir->item_buffer)[ir->read_position];
    (ir->read_position)++;

    if (ir->read_position >= kItemRepositorySize)
        ir->read_position = 0;

    (ir->repo_not_full).notify_all();
    lock.unlock();

    return data;
}

void ProducerTask()
{
    bool ready_to_exit = false;
    while(1) {
        sleep(1);
        std::unique_lock<std::mutex> lock(gItemRepository.item_counter_mtx);
        if (gItemRepository.item_counter < kItemsToProduce) {
            ++(gItemRepository.item_counter);
            ProduceItem(&gItemRepository, gItemRepository.item_counter);
            std::cout << "Producer thread " << std::this_thread::get_id()
                << " is producing the " << gItemRepository.item_counter
                << "^th item" << std::endl;
        } else ready_to_exit = true;
        lock.unlock();
        if (ready_to_exit == true) break;
    }
    std::cout << "Producer thread " << std::this_thread::get_id()
                << " is exiting..." << std::endl;
}

void ConsumerTask()
{
    static int item_consumed = 0;
    while(1) {
        sleep(1);
        ++item_consumed;
        if (item_consumed <= kItemsToProduce) {
            int item = ConsumeItem(&gItemRepository);
            std::cout << "Consumer thread " << std::this_thread::get_id()
                << " is consuming the " << item << "^th item" << std::endl;
        } else break;
    }
    std::cout << "Consumer thread " << std::this_thread::get_id()
                << " is exiting..." << std::endl;
}

void InitItemRepository(ItemRepository *ir)
{
    ir->write_position = 0;
    ir->read_position = 0;
    ir->item_counter = 0;
}

int main()
{
    InitItemRepository(&gItemRepository);
    std::thread producer1(ProducerTask);
    std::thread producer2(ProducerTask);
    std::thread producer3(ProducerTask);
    std::thread producer4(ProducerTask);
    std::thread consumer(ConsumerTask);

    producer1.join();
    producer2.join();
    producer3.join();
    producer4.join();
    consumer.join();
}

多生產者-多消費者模型

該模型可以說是前面兩種模型的綜合，程式需要維護兩個計數器，分別是生產者已生產產品的數目和消費者已取走產品的數目。另外也需要保護產品庫在多個生產者和多個消費者互斥地訪問。

程式碼如下：

#include <unistd.h>

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

static const int kItemRepositorySize  = 4; // Item buffer size.
static const int kItemsToProduce  = 10;   // How many items we plan to produce.

struct ItemRepository {
    int item_buffer[kItemRepositorySize];
    size_t read_position;
    size_t write_position;
    size_t produced_item_counter;
    size_t consumed_item_counter;
    std::mutex mtx;
    std::mutex produced_item_counter_mtx;
    std::mutex consumed_item_counter_mtx;
    std::condition_variable repo_not_full;
    std::condition_variable repo_not_empty;
} gItemRepository;

typedef struct ItemRepository ItemRepository;


void ProduceItem(ItemRepository *ir, int item)
{
    std::unique_lock<std::mutex> lock(ir->mtx);
    while(((ir->write_position + 1) % kItemRepositorySize)
        == ir->read_position) { // item buffer is full, just wait here.
        std::cout << "Producer is waiting for an empty slot...\n";
        (ir->repo_not_full).wait(lock);
    }

    (ir->item_buffer)[ir->write_position] = item;
    (ir->write_position)++;

    if (ir->write_position == kItemRepositorySize)
        ir->write_position = 0;

    (ir->repo_not_empty).notify_all();
    lock.unlock();
}

int ConsumeItem(ItemRepository *ir)
{
    int data;
    std::unique_lock<std::mutex> lock(ir->mtx);
    // item buffer is empty, just wait here.
    while(ir->write_position == ir->read_position) {
        std::cout << "Consumer is waiting for items...\n";
        (ir->repo_not_empty).wait(lock);
    }

    data = (ir->item_buffer)[ir->read_position];
    (ir->read_position)++;

    if (ir->read_position >= kItemRepositorySize)
        ir->read_position = 0;

    (ir->repo_not_full).notify_all();
    lock.unlock();

    return data;
}

void ProducerTask()
{
    bool ready_to_exit = false;
    while(1) {
        sleep(1);
        std::unique_lock<std::mutex> lock(gItemRepository.produced_item_counter_mtx);
        if (gItemRepository.produced_item_counter < kItemsToProduce) {
            ++(gItemRepository.produced_item_counter);
            ProduceItem(&gItemRepository, gItemRepository.produced_item_counter);
            std::cout << "Producer thread " << std::this_thread::get_id()
                << " is producing the " << gItemRepository.produced_item_counter
                << "^th item" << std::endl;
        } else ready_to_exit = true;
        lock.unlock();
        if (ready_to_exit == true) break;
    }
    std::cout << "Producer thread " << std::this_thread::get_id()
                << " is exiting..." << std::endl;
}

void ConsumerTask()
{
    bool ready_to_exit = false;
    while(1) {
        sleep(1);
        std::unique_lock<std::mutex> lock(gItemRepository.consumed_item_counter_mtx);
        if (gItemRepository.consumed_item_counter < kItemsToProduce) {
            int item = ConsumeItem(&gItemRepository);
            ++(gItemRepository.consumed_item_counter);
            std::cout << "Consumer thread " << std::this_thread::get_id()
                << " is consuming the " << item << "^th item" << std::endl;
        } else ready_to_exit = true;
        lock.unlock();
        if (ready_to_exit == true) break;
    }
    std::cout << "Consumer thread " << std::this_thread::get_id()
                << " is exiting..." << std::endl;
}

void InitItemRepository(ItemRepository *ir)
{
    ir->write_position = 0;
    ir->read_position = 0;
    ir->produced_item_counter = 0;
    ir->consumed_item_counter = 0;
}

int main()
{
    InitItemRepository(&gItemRepository);
    std::thread producer1(ProducerTask);
    std::thread producer2(ProducerTask);
    std::thread producer3(ProducerTask);
    std::thread producer4(ProducerTask);

    std::thread consumer1(ConsumerTask);
    std::thread consumer2(ConsumerTask);
    std::thread consumer3(ConsumerTask);
    std::thread consumer4(ConsumerTask);

    producer1.join();
    producer2.join();
    producer3.join();
    producer4.join();

    consumer1.join();
    consumer2.join();
    consumer3.join();
    consumer4.join();
}

相關推薦

C++11 併發指南九(綜合運用: C++11 多執行緒下生產者消費者模型詳解)

前面八章介紹了 C++11 併發程式設計的基礎(抱歉哈，第五章-第八章還在草稿中)，本文將綜合運用 C++11 中的新的基礎設施(主要是多執行緒、鎖、條件變數)來闡述一個經典問題——生產者消費者模型，並給出完整的解決方案。 生產者消費者問題是多執行緒併發中一個非常經典的問題，相信學過作業系統課程的同學都清楚

C++11 多執行緒下生產者消費者模型詳解

 前面八章介紹了 C++11 併發程式設計的基礎(抱歉哈，第五章-第八章還在草稿中)，本文將綜合運用 C++11 中的新的基礎設施(主要是多執行緒、鎖、條件變數)來闡述一個經典問題——生產者消費者模型，並給出完整的解決方案。 生產者消費者問題是多執行緒併發中一個非常

綜合運用: C++11 多線程下生產者消費者模型詳解（轉）

並發 rep 生產 我會 交流 模型 操作 const ref 生產者消費者問題是多線程並發中一個非常經典的問題，相信學過操作系統課程的同學都清楚這個問題的根源。本文將就四種情況分析並介紹生產者和消費者問題，它們分別是：單生產者-單消費者模型，單生產者-多消費者模型，多生產

C++ 11 多執行緒下std::unique_lock與std::lock_guard的區別和用法

這裡主要介紹std::unique_lock與std::lock_guard的區別用法 先說簡單的 一、std::lock_guard的用法 std::lock_guard其實就是簡單的RAII封裝，在建構函式中進行加鎖，解構函式中進行解鎖，這樣可以保證函式退出時，鎖一定被釋放。 簡單來說，就是防止開

JAVA多執行緒--實現生產者/消費者模型：一對一

要實現的功能：執行緒A將一個數據放入佇列，然後通知B進行處理，B處理完後再通知A放一個數據入佇列。 package javathreadlearn; import java.util.LinkedList; import java.util.List; import java.util.Map;

Spring 多執行緒下注入bean問題詳解

本文介紹了Spring 多執行緒下注入bean問題詳解，分享給大家，具體如下： 問題 Spring中多執行緒注入userThreadService注不進去，顯示userThreadService為null異常 程式碼如下： public class UserThreadTask implements

c++11多執行緒 生產者-消費者模型/條件變數condition_variable

1.生產者消費者模型: 在工作中，大家可能會碰到這樣一種情況：某個模組負責產生資料，這些資料由另一個模組來負責處理（此處的模組是廣義的，可以是類、函式、執行緒、程序等）。產生資料的 模組，就形象地稱為生產者；而處理資料的模組，就稱為消費者。在生產者與消費者之間在加個緩衝區

c++11多執行緒中的condition_variable（生產者消費者示例）

#include <iostream> #include <string> #include <th

Linux下多執行緒模擬生產者/消費者問題

/*用執行緒的同步和互斥來實現"生產者-消費者"問題.*/ /* 多生產者多消費者多緩衝區 生產者和消費者不可同時進行 */ #include <stdio.h> #include <stdlib.h> //#include <unistd.h

G++下如何編譯執行C++11多執行緒程式碼

安裝G++的指令：1. 編輯源程式：sudo apt-get install g++    vim hello.cpp#include <iostream>#include <thre

[轉]c++11 多執行緒 future/promise

[轉自 https://blog.csdn.net/jiange_zh/article/details/51602938] 1. < future >標頭檔案簡介 Classes std::future std::future_error std::packaged_task std::pro

c++11多執行緒 thread

 1.thread建構函式 default (1) thread() noexcept; initialization (2) template <class Fn, class... Args> explicit

多執行緒，生產者消費者模式經典講解，簡單易懂2

本模式以一個經典練習為案例： 使用2種鎖機制實現生產者和消費者模式 要求 練習（生產者消費者模式）： 自定義同步容器，容器容量上限為10。可以在多執行緒中應用，並保證資料執行緒安全。 使用synchronized同步及wait()和notifyAll() 實現生產者消費者模式 邏

C++11 多執行緒執行緒共享資料

共享資料的問題 這些在作業系統中都有詳細的介紹，可以回顧作業系統課程。。很典型的就是資料競爭問題。 互斥量保護資料 最原始的方式：使用std::mutex建立互斥量，使用成員lock()加鎖，使用成員unlock()解鎖。但是這種方式需要我們在每個函數出口都呼叫一次unloc

OS:(Linux)多執行緒實現生產者-消費者問題--pthread庫

OS實驗——多執行緒實現生產者-消費者問題時，正確輸入程式碼生成檔案pthread.c，在終端執行： gcc編譯：輸入gcc -o pthread pthread.c 無法成功編譯，錯誤提示如下：   CSDN查詢解決方案後，發現pthread庫並不是Linux系統預設的庫，連結時需要使

c++11多執行緒與執行緒池

最近需要開發一個高效能運算庫，涉及到c++多執行緒的應用，上次做類似的事情已經是4年多以前了，印象中還頗有些麻煩。悔當初做了就算了，也沒想著留點記錄什麼的。這次又研究了一番，發現用上c++11特性之後，現在已經比較簡單了，在此記錄一下。   最簡單的多執行緒情況，不涉及公共變數，各個執行緒之間獨

java多執行緒 kfc 生產者消費者問題

package threadpractice.practice2; public class KFC{ volatile static int num = 0; public KFC() { } public void producer() { synchronized (this) { while

C++11多執行緒（1）

        C++11中添加了duox多執行緒類，編寫C++程式可以直接使用C++11中的多執行緒庫，不必依賴於平臺多執行緒，這樣可以方便寫出誇平臺的多執行緒程式。多執行緒可以最大化利用計算機資源，提高程式碼的執行效率。         C++11中thread類提供兩

C++11多執行緒------std::async

std::async可以認為是封裝了一個std::promise，該函式返回一個std::future，用於獲取其他執行緒的資料。 一般有兩種模式： std::lanch::async：最常用的非同步模式，每次都要執行一遍 std::lanch::defer：只在第

C++11多執行緒---互斥量、鎖、條件變數的總結

關於互斥量std::mutex的總結 互斥量用於組成程式碼的臨界區。C++的多執行緒模型是基於記憶體的，或者說是基於程式碼片段的，這和我們作業系統學習的臨界區概念基本一致，但是與Golang不同，Golang是基於訊息模型的。 一個std::mutex的lock()和unlock