0.關於
為縮短篇幅，本系列記錄如下：
再談多線程模型之生產者消費者（基礎概念）（c++11實現）
再談多線程模型之生產者消費者（單一生產者和單一消費者）（c++11實現）
再談多線程模型之生產者消費者（單一生產者和多消費者）（c++11實現）
再談多線程模型之生產者消費者（多生產者和單一消費者 ）（c++11實現）【本文】
再談多線程模型之生產者消費者（多生產者和多消費者 ）（c++11實現）
再談多線程模型之生產者消費者（總結）（c++11實現）
本文涉及到的代碼演示環境： VS2017
歡迎留言指正
1.多生產者&單一消費者
-
1.1 與 單一生產者和多消費者模型類似， 因為存在多個生產者，需要考慮生產者之間的互斥訪問; 消費者只有一個，因此不存在消費者之間的互斥與競爭。
-
1.2 多個生產者， 可能同時放入商品，類比吃水果，父母同時向果盤放入水果，只有子女中的一個吃水果的情況。
-
1.3 具體點
情況 處理 生產者速率 > 消費者速率 消費者只有一個，因此，不存在消費者之間的競爭。生產者存在多個，多個生產者之間生產好數據就需要按照競爭將數據放入緩沖區，誰先拿到鎖，誰就先放入。最開始，生產者有多個，只能通過競爭生產。但是由於生產效率大於消費速率， 所以定然會出現商品數量 > 消費者數量。當商品總量達到總數，則需要暫停生產，等待消費者消費 生產者速率 < 消費者速率 最開始，剩余放入總數 > 生產者總數，可以同時放入，隨着時間的推移，可能會出現： 剩余放入空間 > 生產者總數 和 剩余放入空間 < 生產者總數。 當出現 剩余空間 < 生產者總數 時，已經不滿足同時放入，此時就需要鎖。來保證 -
1.4 結構體模型
template<typename T>
struct repo_
{
// 用作互斥訪問緩沖區
std::mutex _mtx_queue;
// 緩沖區最大size
unsigned int _count_max_queue_10 = 10;
// 緩沖區
std::queue<T> _queue;
// 緩沖區沒有滿，通知生產者繼續生產
std::condition_variable _cv_queue_not_full;
// 緩沖區不為空，通知消費者繼續消費
std::condition_variable _cv_queue_not_empty;
// 用於生產者之間的競爭
std::mutex _mtx_pro;
// 計算當前已經生產了多少數據了
unsigned int _cnt_cur_pro = 0;
repo_(const unsigned int count_max_queue = 10) :_count_max_queue_10(count_max_queue)
, _cnt_cur_con(0)
{
;
}
repo_(const repo_&instance) = delete;
repo_& operator = (const repo_& instance) = delete;
repo_(const repo_&&instance) = delete;
repo_& operator = (const repo_&& instance) = delete;
};
對比單一生產者和單一消費者 可知，僅僅增加了下面的代碼
// 用於生產者之間的競爭
std::mutex _mtx_pro;
// 計算當前已經生產了多少數據了
unsigned int _cnt_cur_pro = 0;
- 1.5 生產者線程變化如下
template< typename T >
void thread_pro(const int thread_index, const int count_max_produce, repo<T>* param_repo)
{
if (nullptr == param_repo || NULL == param_repo)
return;
while (true)
{
bool is_running = true;
{
// 用於生產者之間競爭
std::unique_lock<std::mutex> lock(param_repo->_mtx_pro);
// 緩沖區沒有滿，繼續生產
if (param_repo->_cnt_cur_pro < cnt_total_10)
{
thread_produce_item<T>(thread_index, *param_repo, param_repo->_cnt_cur_pro);
++param_repo->_cnt_cur_pro;
}
else
is_running = false;
}
std::this_thread::sleep_for(std::chrono::microseconds(16));
if (!is_running)
break;
}
}
- 1.6 完整源碼
#pragma once
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <vector>
std::mutex _mtx;
std::condition_variable _cv_not_full;
std::condition_variable _cv_not_empty;
const int max_queue_size_10 = 10;
enum
{
// 總生產數目
cnt_total_10 = 10,
};
template<typename T>
struct repo_
{
// 用作互斥訪問緩沖區
std::mutex _mtx_queue;
// 緩沖區最大size
unsigned int _count_max_queue_10 = 10;
// 緩沖區
std::queue<T> _queue;
// 緩沖區沒有滿，通知生產者繼續生產
std::condition_variable _cv_queue_not_full;
// 緩沖區不為空，通知消費者繼續消費
std::condition_variable _cv_queue_not_empty;
// 用於生產者之間的競爭
std::mutex _mtx_pro;
// 計算當前已經生產了多少數據了
unsigned int _cnt_cur_pro = 0;
repo_(const unsigned int count_max_queue = 10) :_count_max_queue_10(count_max_queue)
, _cnt_cur_con(0)
{
;
}
repo_(const repo_&instance) = delete;
repo_& operator = (const repo_& instance) = delete;
repo_(const repo_&&instance) = delete;
repo_& operator = (const repo_&& instance) = delete;
};
template <typename T>
using repo = repo_<T>;
//----------------------------------------------------------------------------------------
// 生產者生產數據
template <typename T>
void thread_produce_item(const int &thread_index, repo<T>& param_repo, const T& repo_item)
{
std::unique_lock<std::mutex> lock(param_repo._mtx_queue);
// 1. 生產者只要發現緩沖區沒有滿， 就繼續生產
param_repo._cv_queue_not_full.wait(lock, [&] { return param_repo._queue.size() < param_repo._count_max_queue_10; });
// 2. 將生產好的商品放入緩沖區
param_repo._queue.push(repo_item);
// log to console
std::cout << "生產者" << thread_index << "生產數據：" << repo_item << "\n";
// 3. 通知消費者可以消費了
//param_repo._cv_queue_not_empty.notify_one();
param_repo._cv_queue_not_empty.notify_one();
}
//----------------------------------------------------------------------------------------
// 消費者消費數據
template <typename T>
T thread_consume_item(const int thread_index, repo<T>& param_repo)
{
std::unique_lock<std::mutex> lock(param_repo._mtx_queue);
// 1. 消費者需要等待【緩沖區不為空】的信號
param_repo._cv_queue_not_empty.wait(lock, [&] {return !param_repo._queue.empty(); });
// 2. 拿出數據
T item;
item = param_repo._queue.front();
param_repo._queue.pop();
std::cout << "消費者" << thread_index << "從緩沖區中拿出一組數據:" << item << std::endl;
// 3. 通知生產者，繼續生產
param_repo._cv_queue_not_full.notify_one();
return item;
}
//----------------------------------------------------------------------------------------
/**
* @ brief: 生產者線程
* @ thread_index - 線程標識，區分是哪一個線程
* @ count_max_produce - 最大生產次數
* @ param_repo - 緩沖區
* @ return - void
*/
template< typename T >
void thread_pro(const int thread_index, const int count_max_produce, repo<T>* param_repo)
{
if (nullptr == param_repo || NULL == param_repo)
return;
while (true)
{
bool is_running = true;
{
// 用於生產者之間競爭
std::unique_lock<std::mutex> lock(param_repo->_mtx_pro);
// 緩沖區沒有滿，繼續生產
if (param_repo->_cnt_cur_pro < cnt_total_10)
{
thread_produce_item<T>(thread_index, *param_repo, param_repo->_cnt_cur_pro);
++param_repo->_cnt_cur_pro;
}
else
is_running = false;
}
std::this_thread::sleep_for(std::chrono::microseconds(16));
if (!is_running)
break;
}
}
/**
* @ brief: 消費者線程
* @ thread_index - 線程標識，區分線程
* @ param_repo - 緩沖區
* @ return - void
*/
template< typename T >
void thread_con(const int thread_index, repo<T>* param_repo)
{
if (nullptr == param_repo || NULL == param_repo)
return;
static int cnt_cur_con = 0;
while (true)
{
bool is_running = true;
{
// std::unique_lock<std::mutex> lock(param_repo->_mtx_con);
// 還沒消費到指定的數目，繼續消費
if (cnt_cur_con < cnt_total_10)
{
thread_consume_item<T>(thread_index, *param_repo);
++cnt_cur_con;
}
else
is_running = false;
}
std::this_thread::sleep_for(std::chrono::microseconds(16));
// 結束線程
if ((!is_running))
break;
}
}
// 入口函數
//----------------------------------------------------------------------------------------
int main(int argc, char *argv[], char *env[])
{
// 緩沖區
repo<int> repository;
// 線程池
std::vector<std::thread> vec_thread;
// 生產者
vec_thread.push_back(std::thread(thread_pro<int>, 1, cnt_total_10, &repository));
vec_thread.push_back(std::thread(thread_pro<int>, 2, cnt_total_10, &repository));
// 消費者
vec_thread.push_back(std::thread(thread_con<int>, 1, &repository));
for (auto &item : vec_thread)
{
item.join();
}
return 0;
}
- 1.7 可能結果
