什么是生产者消费者模型
生产者消费者模型是多线程中一个比较典型的模型。
打个比方：你是一个客户，你去超市里买火腿肠。
这段话中的 "你"就是消费者， 那么给超市提供火腿肠的供货商就是生产者。超市呢？超市是不是被所有人所共享？大家都可以去访问超市，所以这里的超市是一份临界资源。
所以生产者消费者有三种关系，两种角色，一个交易场所。
三种关系：
1.生产者与生产者
2.消费者与消费者
3.生产者与消费者
生产者与生产者是竞争关系，因为厂商之间互相竞争。所以生产与生产者是互斥关系。
消费者与消费者其实也是竞争关系，但是因为商品够多，而消费者消费速度太慢，所以没有明显的区别。但如果世界上只剩下最后一瓶矿泉水了，那是不是大家都会去抢呢？ 所以消费者与消费者其实也是互斥关系。
生产者与消费者也是竞争关系，我们生产者和消费者看成两个线程，超市看成一份临界资源。那么这两个线程是不是都要访问这个临界资源？既然都要访问这个临界资源，那么生产和消费者也是互斥关系。但不仅仅是互斥，因为生产者把超市装满了，是不是要等待用户来消费？同理如果超市空了，消费者是不是要等待生产者来供货？所以生产和消费者还有一层关系，那就是同步。
两种角色
生产者与消费者
一个交易场所
一份临界资源，生产者向临界资源提供数据，消费者从临界资源中拿数据。
有没有发现生产与消费者模型很像管道？没错，管道就是典型的生产者与消费者模型。
这是一个多生产者多消费者的模型。
接下来我们就来实现一个基于阻塞队列的生产者消费者模型。这里的阻塞队列冲当的就是临界资源，生产者把数据放进阻塞队列，消费者把数据从阻塞队列中拿出。
锁的封装
首先我们用RAII风格的锁。
MyLock类
#include<pthread.h>
class MyLock
{
public:
MyLock(pthread_mutex_t* pmtx): _pmtx(pmtx){}
void Lock(){ pthread_mutex_lock(_pmtx);}
void Unlock() { pthread_mutex_unlock(_pmtx);}
private:
pthread_mutex_t* _pmtx;
};
LockGuard类
#include<pthread.h>
class LockGuard
{
public:
LockGuard(pthread_mutex_t* pmtx):_mtx(pmtx){
_mtx.Lock();
}
~LockGuard()
{
_mtx.Unlock();
}
private:
MyLock _mtx;
};
这个类的构造函数是加锁，析构函数是解锁。所以我们只需要创建一个这个类的对象的代码和临界资源的代码放在一起，就可以实现加锁和解锁了。这种方式可以避免有时候解锁忘记写了导致死锁的问题。
阻塞队列的实现
block_queue类的声明
#include<queue>
#include<pthread.h>
#include<iostream>
#include "Task.hpp"
#include "LockGuard.hpp"
#define DEFAULT_NUM 5
template<class T> //因为不确定阻塞队列放的数据类型， 所以用模板参数
class block_queue
{
private:
size_t _num; //阻塞队列的容量
std::queue<T> _blockqueue; //阻塞队列
pthread_mutex_t _mtx; //锁
pthread_cond_t _full; //条件变量，让生产者在阻塞队列为满时进行等待
pthread_cond_t _empty; //条件变量，让消费者在阻塞队列为空时进行等待
public:
block_queue(size_t num = DEFAULT_NUM); //构造函数
~block_queue(); // 析构
//生产者生产
void Push(const T& task);
// 消费者消费
void Pop(T* out);
private:
//让当前线程在指定的条件变量下等待
void Wait(pthread_cond_t* cond) {pthread_cond_wait(cond,&_mtx);}
//唤醒指定条件变量下等待的线程
void Wakeup(pthread_cond_t* cond) {pthread_cond_signal(cond);}
//判断阻塞队列是否满了
bool isfull() { return _blockqueue.size() == _num;}
//判断阻塞队列是否为空
bool isempty() { return _blockqueue.size() == 0;}
};
我们的阻塞队列实际上只提供2个操作，一个是push(生产者放数据)，一个是pop(消费者拿数据)。
block_queue类的实现
#define DEFAULT_NUM 5
template<class T>
class block_queue
{
private:
size_t _num;
std::queue<T> _blockqueue;
pthread_mutex_t _mtx;
pthread_cond_t _full;
pthread_cond_t _empty;
public:
block_queue(size_t num = DEFAULT_NUM) : _num(num){
pthread_mutex_init(&_mtx,nullptr);
pthread_cond_init(&_full,nullptr);
pthread_cond_init(&_empty,nullptr);
}
~block_queue()
{
pthread_mutex_destroy(&_mtx);
pthread_cond_destroy(&_full);
pthread_cond_destroy(&_empty);
}
//生产者生产
void Push(const T& task)
{
LockGuard lockguard(&_mtx); //加锁，出了作用域自动解锁
while(isfull()) Wait(&_full); //生产队列已满，生产者在full条件变量下等待
//被唤醒后添加任务到生产队列
_blockqueue.push(task);
printf("%p 生产了一个任务 : %d %c %d\n",pthread_self(),task._x,task._op,task._y); //这是对任务的打印....暂且无视，等Task类实现完后看结果的
Wakeup(&_empty); //唤醒消费者
}
// 消费者消费
void Pop(T* out)
{
LockGuard lockguard(&_mtx) ;//加锁，出了作用域自动解锁
while(isempty()) Wait(&_empty); //生产队列已空，消费者进入等待
//被唤醒后添加任务到生产队列
*out = _blockqueue.front(); //提取任务
_blockqueue.pop(); //队列pop
Wakeup(&_full);
}
private:
void Wait(pthread_cond_t* cond) {pthread_cond_wait(cond,&_mtx);}
void Wakeup(pthread_cond_t* cond) {pthread_cond_signal(cond);}
bool isfull() { return _blockqueue.size() == _num;}
bool isempty() { return _blockqueue.size() == 0;}
};
Task类实现
我们可以往阻塞队列里面放数据，当然也可以往里面放一个任务。这里我们就创建一个加减乘除取模运算的任务类。
#include <iostream>
class Task{
public:
Task(){}
Task(int x, char op,int y):_x(x),_op(op),_y(y),_iserror(false){}
void Runing()
{
int ret = 0;
switch(_op)
{
case '+' : ret = _x + _y; break;
case '-' : ret = _x - _y; break;
case '*' : ret = _x * _y; break;
case '/' :
{
if(_y) ret = _x / _y;
else _iserror = true;
break;
}
case '%' :
{
if(_y) ret = _x % _y;
else _iserror = true;
break;
}
default: _iserror = true;
}
if(_iserror) std::cout << "result error" << std::endl; //如果结果错误打印错误
else std::cout << _x << _op << _y << "=" << ret << std::endl; //如果结果正确打印完整式子
}
public:
int _x; //第一个操作数
char _op; //操作符
int _y; //第二个操作数
bool _iserror; //结果是否错误
};
Main
`
#include "BlockQueue.hpp"
#include <time.h>
#include<unistd.h>
#include<string>
#define CONNUM 5
#define PRODNUM 2
//生产者放任务
void* ProcuderRuning(void* args)
{
wyl::block_queue<wyl::Task>* bq = (wyl::block_queue<wyl::Task>*)args;
while(1)
{
int x = rand() % 10 + 1;
int y = rand()%20;
char op = "+-*/%"[rand() % 5];
bq->Push(wyl::Task(x,op,y)); //往阻塞队列中放任务
}
}
//消费不断拿任务
void* ConsumerRuning(void* args)
{
wyl::block_queue<wyl::Task>* bq = (wyl::block_queue<wyl::Task>*)args;
while(1)
{
wyl::Task t;
bq->Pop(&t); //从阻塞队列中拿任务
printf("%p 消费了一个任务",pthread_self());
t.Runing(); //处理任务
sleep(1); //让消费者不要频繁消费太快，这样阻塞队列满了会等待消费者
}
}
int main()
{
pthread_t con[CONNUM];
pthread_t prod[PRODNUM];
srand((unsigned int)0); //随机数种子
//创造等待队列
wyl::block_queue<wyl::Task>* bq = new wyl::block_queue<wyl::Task>(5);
//创建生产者线程
for(int i = 0 ; i < PRODNUM ; i++)
{
std::string name = "prodcuer ";
name += std::to_string(i+1);
pthread_create(prod + i,nullptr,ProcuderRuning,(void*)bq);
}
//创建消费者线程
for(int i = 0 ; i < CONNUM ; i++)
{
std::string name = "consumer ";
name += std::to_string(i+1);
pthread_create(con + i,nullptr,ConsumerRuning,(void*)bq);
}
//等待线程
for(int i = 0 ; i < PRODNUM ; i++)
{
pthread_join(prod[i],nullptr);
}
for(int i = 0 ; i < CONNUM ; i++)
{
pthread_join(con[i],nullptr);
}
return 0;
}
`
消费者慢消费，生产者快生产的执行结果：
生产者慢生产，消费者快消费的运行结果：
我们会发现，任务井然有序的执行。生产者放了数据后通知消费拿，消费者把数据拿完又会通知生产者放。