一、生產(chǎn)者消費者模型
我們這里舉一個例子，來解釋生產(chǎn)者消費者模型，我們學(xué)生–消費者，供應(yīng)商–生產(chǎn)者，超市–交易場所，我們買東西只需要關(guān)系售貨架子上是否有商品即可，沒有了商品，超市從供應(yīng)商進行供貨。供應(yīng)商和供應(yīng)商不能同時向一個貨架進行供貨，所以生產(chǎn)者之間是互斥的關(guān)系，非消費者和消費不能同時從同一個貨架拿商品，所以消費者與消費者之間是互斥的關(guān)系，而消費者需要等生產(chǎn)者將商品放到貨架之后才能拿取商品，所以生產(chǎn)者和消費者之間是互斥和同步的關(guān)系。
生產(chǎn)消費模型:
生產(chǎn)者和生產(chǎn)者之間:互斥關(guān)系
消費者和消費者之間:互斥關(guān)系
生產(chǎn)者和消費者之間:互斥&&同步
總結(jié):“321”原則:
3種關(guān)系:生產(chǎn)者和生產(chǎn)者（互斥)，消費者和消費者（互斥)，生產(chǎn)者和消費者
互斥保證共享資源的安全性]同步)–產(chǎn)品（數(shù)據(jù))
種角色:生產(chǎn)者線程，消費者線程
1個交易場所:(一段特定結(jié)構(gòu)的緩沖區(qū)
只要我們想寫生產(chǎn)消費模型，我們本質(zhì)工作其實就是維護321原則!
挖掘特點:
1.生產(chǎn)線程和消費線程進行解耦
2支持生產(chǎn)和消費的一段時間的忙閑不均的問題
3.提高效率
生產(chǎn)者消費者模式就是通過一個容器來解決生產(chǎn)者和消費者的強耦合問題。生產(chǎn)者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊列里取，阻塞隊列就相當(dāng)于一個緩沖區(qū)，平衡了生產(chǎn)者和消費者的處理能力。這個阻塞隊列就是用來給生產(chǎn)者和消費者解耦的。
生產(chǎn)者消費者模型優(yōu)點
解耦
支持并發(fā)
支持忙閑不均
二、基于BlockingQueue的生產(chǎn)者消費者模型
BlockingQueue
在多線程編程中阻塞隊列(Blocking Queue)是一種常用于實現(xiàn)生產(chǎn)者和消費者模型的數(shù)據(jù)結(jié)構(gòu)。其與普通的隊列區(qū)別在于，當(dāng)隊列為空時，從隊列獲取元素的操作將會被阻塞，直到隊列中被放入了元素；當(dāng)隊列滿時，往隊列里存放元素的操作也會被阻塞，直到有元素被從隊列中取出(以上的操作都是基于不同的線程來說的，線程在對阻塞隊列進程操作時會被阻塞)
1.BlockQueue.hpp
#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
using namespace std;
const int gnum = 5;
template <class T>
class BlockQueue
{
public:
BlockQueue(const int& maxcap = gnum)
: _maxcap(maxcap)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_pcond, nullptr);
pthread_cond_init(&_ccond, nullptr);
}
void push(const T &in) // 輸出型參數(shù)：*， // 輸入輸出型：&
{
pthread_mutex_lock(&_mutex);
// 1. 判斷
// 細(xì)節(jié)2: 充當(dāng)條件判斷的語法必須是while，不能用if
while (is_full())
{
// 細(xì)節(jié)1：pthread_cond_wait這個函數(shù)的第二個參數(shù)，必須是我們正在使用的互斥鎖！
// a. pthread_cond_wait: 該函數(shù)調(diào)用的時候，會以原子性的方式，將鎖釋放，并將自己掛起
// b. pthread_cond_wait: 該函數(shù)在被喚醒返回的時候，會自動的重新獲取你傳入的鎖
pthread_cond_wait(&_pcond, &_mutex);
}
// 2. 走到這里一定是沒有滿
_q.push(in);
// 3. 絕對能保證，阻塞隊列里面一定有數(shù)據(jù)
// 細(xì)節(jié)3：pthread_cond_signal：這個函數(shù)，可以放在臨界區(qū)內(nèi)部，也可以放在外部
pthread_cond_signal(&_ccond);
pthread_mutex_unlock(&_mutex);
}
void pop(T *out)
{
pthread_mutex_lock(&_mutex);
// 1. 判斷
while (is_empty())
{
pthread_cond_wait(&_ccond, &_mutex);
}
// 2. 走到這里我們能保證，一定不為空
*out = _q.front();
_q.pop();
// 3. 絕對能保證，阻塞隊列里面，至少有一個空的位置！
pthread_cond_signal(&_pcond);
pthread_mutex_unlock(&_mutex);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_pcond);
pthread_cond_destroy(&_ccond);
}
private:
bool is_full()
{
return _q.size() == _maxcap;
}
bool is_empty()
{
return _q.empty();
}
private:
queue<T> _q;
int _maxcap; // 隊列中元素的上限
pthread_mutex_t _mutex;
pthread_cond_t _pcond; // 生產(chǎn)者對應(yīng)的條件變量
pthread_cond_t _ccond; // 消費者對應(yīng)的條件變量
};
2.Task.hpp
#pragma once
#include <iostream>
#include <string>
#include <functional>
class CalTask
{
public:
typedef std::function<int(int, int, char)> func_t;
// using func_t = std::function<int(int, int, char)>;
public:
CalTask()
{
}
CalTask(int x, int y, char op, func_t func)
: _x(x), _y(y), _op(op), _callback(func)
{
}
std::string operator()()
{
int result = _callback(_x, _y, _op);
char buffer[1024];
snprintf(buffer, sizeof buffer, "%d %c %d = %d", _x, _op, _y, result);
return buffer;
}
std::string toTaskString()
{
char buffer[1024];
snprintf(buffer, sizeof buffer, "%d %c %d = ?", _x, _op, _y);
return buffer;
}
private:
int _x;
int _y;
char _op;
func_t _callback;
};
const std::string oper = "+-*/%";
int calculate(int x, int y, char op)
{
int result = 0;
switch (op)
{
case '+':
result = x + y;
break;
case '-':
result = x - y;
break;
case '*':
result = x * y;
break;
case '/':
{
if (y == 0)
{
std::cerr << "div zero error" << std::endl;
return -1;
}
else
result = x / y;
}
break;
case '%':
{
if (y == 0)
{
std::cerr << "mod zero error" << std::endl;
return -1;
}
else
result = x % y;
}
break;
default:
std::cerr << "請輸入正確的操作符" << std::endl;
break;
}
return result;
}
class SaveTask
{
public:
using func_t = function<void(const std::string &)>;
public:
SaveTask()
{
}
SaveTask(const std::string &message, func_t func)
: _message(message), _func(func)
{
}
void operator()()
{
_func(_message);
}
~SaveTask()
{
}
private:
std::string _message;
func_t _func;
};
void Save(const std::string &message)
{
const std::string target = "./log.txt";
FILE *fp = fopen(target.c_str(), "a+");
if (fp == nullptr)
{
perror("fopen error");
return;
}
fputs(message.c_str(), fp);
fputs("\n", fp);
fclose(fp);
}
3.main.cc
#include "BlockQueue.hpp"
#include "Task.hpp"
#include <ctime>
#include <unistd.h>
#include <sys/types.h>
// C：計算
// S: 存儲
template <class C, class S>
class BlockQueues
{
public:
BlockQueue<C> *_cbq;
BlockQueue<S> *_sbq;
};
void *productor(void *args)
{
BlockQueue<CalTask> *bq = static_cast<BlockQueues<CalTask, SaveTask> *>(args)->_cbq;
while (true)
{
int x = rand() % 10 + 1;
int y = rand() % 5;
char op = oper[rand() % 5];
CalTask t(x, y, op, calculate);
bq->push(t);
std::cout << "productor thread 生產(chǎn)計算任務(wù): " << t.toTaskString() << std::endl;
sleep(1);
}
return nullptr;
}
void *consumer(void *args)
{
BlockQueue<CalTask> *bq = static_cast<BlockQueues<CalTask, SaveTask> *>(args)->_cbq;
BlockQueue<SaveTask> *save_bq = static_cast<BlockQueues<CalTask, SaveTask> *>(args)->_sbq;
while (true)
{
CalTask t;
bq->pop(&t);
std::string result = t();
std::cout << "cal thread,完成計算任務(wù): " << result << " ... done" << std::endl;
SaveTask st(result, Save);
save_bq->push(st);
cout << "cal thread,推送存儲任務(wù)完成..." << std::endl;
sleep(1);
}
return nullptr;
}
void *saver(void *args)
{
BlockQueue<SaveTask> *save_bq = static_cast<BlockQueues<CalTask, SaveTask> *>(args)->_sbq;
while (true)
{
SaveTask st;
save_bq->pop(&st);
st();
std::cout << "save thread,保存任務(wù)完成..." << std::endl;
// sleep(1);
}
return nullptr;
}
int main()
{
BlockQueues<CalTask, SaveTask> bqs;
bqs._cbq = new BlockQueue<CalTask>();
bqs._sbq = new BlockQueue<SaveTask>();
pthread_t p, c, s;
pthread_create(&p, nullptr, productor, &bqs);
pthread_create(&c, nullptr, consumer, &bqs);
pthread_create(&s, nullptr, saver, &bqs);
pthread_join(p, nullptr);
pthread_join(c, nullptr);
pthread_join(s, nullptr);
delete bqs._cbq;
delete bqs._sbq;
return 0;
}
你創(chuàng)建多線程生產(chǎn)和消費的意義是什么??2.生產(chǎn)消費模型高效在哪里??
可以在生產(chǎn)之前，和消費之后，讓線程并行執(zhí)行
生產(chǎn)者而言，向blockqueue里面放置任務(wù)
他的任務(wù)從哪里來的呢?它獲取任務(wù)和構(gòu)建任務(wù)要不要花時間？
消費者而言，從blockqueue里面拿取任務(wù)
對于消費者，難道他把任務(wù)從任務(wù)隊列中拿出來就完了嗎??消費者拿到任務(wù)之后，后續(xù)還有沒有任務(wù)??
三、POSIX信號量
先發(fā)現(xiàn)我們之前寫的代碼的不足的地方
pthread_mutex_lock(&_mutex);
while (is_full())
{
pthread_cond_wait(&_pcond, &_mutex);
}
_q.push(in);
pthread_cond_signal(&_ccond);
pthread_mutex_unlock(&_mutex);
1.一個線程，在操作臨界資源的時候，必須臨界資源是滿足條件的!
2.可是，公共資源是否滿足生產(chǎn)或者消費條件，我們無法、直接得知【我們不能事前得知【在沒有訪問之前，無法得知】】
3.只能先加鎖，再檢測，再操作，再解鎖。因為你要檢測，本質(zhì):也是在訪問臨界資源!
因為我們在操作臨界資源的時候，有可能不就緒但是，我們無法提前得知，所以，只能先加鎖，在檢測，根據(jù)檢測結(jié)果，決定下一步怎么走!
只要我們對資源進行整體加鎖，就默認(rèn)了，我們對這個資源整體使用。實際情況可能存在:.一份公共資源，但是允許同時訪問不同的區(qū)域!程序員編碼保證不同的線程可以并發(fā)訪問公共資源的不同區(qū)域!
什么是信號量:
a.信號量本質(zhì)是一把計數(shù)器。衡量臨界資源中資源數(shù)量多少的計數(shù)器
b.只要擁有信號量，就在未來一定能夠擁有臨界資源的一部分，申請信號量的本質(zhì):對臨界資源中特定小塊資源的預(yù)訂機制
線程要訪問臨界資源中的某一區(qū)域–申請信號量–所有人必須的先看到信號量–信號量本身必須是:公共資源
有可能，我們在訪問真正的臨界資源之前，我們其實就可以提前知道臨界資源的使用情況! ! !
信號量只要申請成功，就一定有你的資源。只要申請失敗，就說明條件不就緒，你只能等!!不需要在判斷了!
計數(shù)器–遞減or 遞增sem_t sem = 10;
sem–;—申請資源—必須保證操作的原子性— P
sem++;—歸還資源—必須保證操作的原子性----V
信號量核心操作:PV原語
POSIX信號量和SystemV信號量作用相同，都是用于同步操作，達(dá)到無沖突的訪問共享資源目的。 但POSIX可以用于線程間同步。
初始化信號量
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
參數(shù)：
pshared:0表示線程間共享，非零表示進程間共享
value：信號量初始值
銷毀信號量
int sem_destroy(sem_t *sem);
等待信號量
功能：等待信號量，會將信號量的值減1
int sem_wait(sem_t *sem); //P()
發(fā)布信號量
功能：發(fā)布信號量，表示資源使用完畢，可以歸還資源了。將信號量值加1。
int sem_post(sem_t *sem);//V()
四、基于環(huán)形隊列的生產(chǎn)消費模型
環(huán)形隊列采用數(shù)組模擬，用模運算來模擬環(huán)狀特性
環(huán)形結(jié)構(gòu)起始狀態(tài)和結(jié)束狀態(tài)都是一樣的，不好判斷為空或者為滿，所以可以通過加計數(shù)器或者標(biāo)記位來判斷滿或者空。另外也可以預(yù)留一個空的位置，作為滿的狀態(tài)
但是我們現(xiàn)在有信號量這個計數(shù)器，就很簡單的進行多線程間的同步過程
生產(chǎn)和消費在什么情況下可能訪問同一個位置:
1.空的時候
2.滿的時候
3.其他情況，生產(chǎn)者和消費者，根本訪問的就是不同的區(qū)域!
為了完成環(huán)形隊列cp問題，我們要做的核心工作是什么
1.你不能超過我
2我不能把你套一個圈以上
3．我們兩個什么時候，會站在一起?
信號量是用來衡量臨界資源中資源數(shù)量的
1.對于生產(chǎn)者而言，看中什么?隊列中的剩余空間—空間資源定義十個信號量
2.對于消費者而言，看中的是什么?放入隊列中的數(shù)據(jù)! —數(shù)據(jù)資源定義一個信號量
生產(chǎn)者而言:
prodocter_sem: 0
// 申請成功，你就可以繼續(xù)向下運行。
//申請失敗，當(dāng)前執(zhí)行流，阻塞在申請?zhí)?/span>
P(producter_sem);
//從事生產(chǎn)活動--把數(shù)據(jù)放入到隊列中
V(comsumer_sem);
消費者而言:
comsumer_sem:10
P(comsumer_sem);
//從事消費活動
v(producter_sem);_
未來，生產(chǎn)和消費的位置我們要想清楚:
1.其實就是隊列中的下標(biāo)、
2一定是兩個下標(biāo)文章來源：http://www.zghlxwxcb.cn/news/detail-768050.html
3.為空或者為滿，下標(biāo)相同文章來源地址http://www.zghlxwxcb.cn/news/detail-768050.html
1.RingQueue.hpp
#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <cassert>
#include <semaphore.h>
static const int gcap = 5;
template <class T>
class RingQueue
{
private:
// 等待信號量，會將信號量的值減1
void P(sem_t &sem)
{
int n = sem_wait(&sem);
assert(n == 0);
(void)n;
}
// 發(fā)布信號量，表示資源使用完畢，可以歸還資源了。將信號量值加1。
void V(sem_t &sem)
{
int n = sem_post(&sem);
assert(n == 0);
(void)n;
}
public:
RingQueue(const int &cap = gcap)
: _queue(cap), _cap(cap)
{
int n = sem_init(&_spaceSem, 0, _cap);
assert(n == 0);
n = sem_init(&_dataSem, 0, 0);
assert(n == 0);
_productorStep = _consumerStep = 0;
pthread_mutex_init(&_pmutex, nullptr);
pthread_mutex_init(&_cmutex, nullptr);
}
// 生產(chǎn)者
void push(const T &in)
{
P(_spaceSem);
pthread_mutex_lock(&_pmutex);
_queue[_productorStep++] = in;
_productorStep %= _cap;
pthread_mutex_unlock(&_pmutex);
V(_dataSem);
}
// 消費者
void pop(T *out)
{
P(_dataSem);
pthread_mutex_lock(&_cmutex);
*out = _queue[_consumerStep++];
_consumerStep %= _cap;
pthread_mutex_unlock(&_cmutex);
V(_spaceSem);
}
~RingQueue()
{
sem_destroy(&_spaceSem);
sem_destroy(&_dataSem);
pthread_mutex_destroy(&_pmutex);
pthread_mutex_destroy(&_cmutex);
}
private:
std::vector<T> _queue;
int _cap;
sem_t _spaceSem; // 生產(chǎn)者 想生產(chǎn)，看中的是什么資源呢? 空間資源
sem_t _dataSem; // 消費者 想消費，看中的是什么資源呢? 數(shù)據(jù)資源
int _productorStep;
int _consumerStep;
pthread_mutex_t _pmutex;
pthread_mutex_t _cmutex;
};
2.Task.hpp
#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <functional>
class Task
{
using func_t = std::function<int(int,int,char)>;
// typedef std::function<int(int,int)> func_t;
public:
Task()
{}
Task(int x, int y, char op, func_t func)
:_x(x), _y(y), _op(op), _callback(func)
{}
std::string operator()()
{
int result = _callback(_x, _y, _op);
char buffer[1024];
snprintf(buffer, sizeof buffer, "%d %c %d = %d", _x, _op, _y, result);
return buffer;
}
std::string toTaskString()
{
char buffer[1024];
snprintf(buffer, sizeof buffer, "%d %c %d = ?", _x, _op, _y);
return buffer;
}
private:
int _x;
int _y;
char _op;
func_t _callback;
};
const std::string oper = "+-*/%";
int mymath(int x, int y, char op)
{
int result = 0;
switch (op)
{
case '+':
result = x + y;
break;
case '-':
result = x - y;
break;
case '*':
result = x * y;
break;
case '/':
{
if (y == 0)
{
std::cerr << "div zero error!" << std::endl;
result = -1;
}
else
result = x / y;
}
break;
case '%':
{
if (y == 0)
{
std::cerr << "mod zero error!" << std::endl;
result = -1;
}
else
result = x % y;
}
break;
default:
// do nothing
break;
}
return result;
}
3.main.cc
#include "RingQueue.hpp"
#include "Task.hpp"
#include <pthread.h>
#include <ctime>
#include <cstdlib>
#include <sys/types.h>
#include <unistd.h>
std::string SelfName()
{
char name[128];
snprintf(name, sizeof(name), "thread[0x%x]", pthread_self());
return name;
}
void *ProductorRoutine(void *rq)
{
// RingQueue<int> *ringqueue = static_cast<RingQueue<int> *>(rq);
RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
while (true)
{
// version1
// int data = rand() % 10 + 1;
// ringqueue->Push(data);
// std::cout << "生產(chǎn)完成，生產(chǎn)的數(shù)據(jù)是：" << data << std::endl;
// version2
// 構(gòu)建or獲取任務(wù) --- 這個是要花時間的！
int x = rand() % 10;
int y = rand() % 5;
char op = oper[rand() % oper.size()];
Task t(x, y, op, mymath);
// 生產(chǎn)任務(wù)
ringqueue->push(t);
// 輸出提示
std::cout << SelfName() << ", 生產(chǎn)者派發(fā)了一個任務(wù): " << t.toTaskString() << std::endl;
// sleep(1);
}
}
void *ConsumerRoutine(void *rq)
{
// RingQueue<int> *ringqueue = static_cast<RingQueue<int> *>(rq);
RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
while (true)
{
// version1
// int data;
// ringqueue->Pop(&data);
// std::cout << "消費完成，消費的數(shù)據(jù)是：" << data << std::endl;
// sleep(1);
// version2
Task t;
// 消費任務(wù)
ringqueue->pop(&t);
std::string result = t(); // 消費也是要花時間的！
std::cout << SelfName() << ", 消費者消費了一個任務(wù): " << result << std::endl;
// sleep(1);
}
}
int main()
{
srand((unsigned int)time(nullptr) ^ getpid() ^ pthread_self() ^ 0x71727374);
// RingQueue<int> *rq = new RingQueue<int>();
RingQueue<Task> *rq = new RingQueue<Task>();
// 單生產(chǎn)，單消費，多生產(chǎn)，多消費 --> 只要保證，最終進入臨界區(qū)的是一個生產(chǎn)，一個消費就行！
// 多生產(chǎn)，多消費的意義？？
pthread_t p[4], c[8];
for (int i = 0; i < 4; i++)
pthread_create(p + i, nullptr, ProductorRoutine, rq);
for (int i = 0; i < 8; i++)
pthread_create(c + i, nullptr, ConsumerRoutine, rq);
for (int i = 0; i < 4; i++)
pthread_join(p[i], nullptr);
for (int i = 0; i < 8; i++)
pthread_join(c[i], nullptr);
delete rq;
return 0;
}
到了這里，關(guān)于【Linux】生產(chǎn)者消費者模型(阻塞隊列與環(huán)形隊列)和POSIX信號量的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
