并发编程之生产者消费者模型

什么是生产者消费者模型

生产者消费者模型是多线程中一个比较典型的模型。

打个比方：你是一个客户，你去超市里买火腿肠。

这段话中的 "你"就是消费者， 那么给超市提供火腿肠的供货商就是生产者。超市呢？超市是不是被所有人所共享？大家都可以去访问超市，所以这里的超市是一份临界资源。

所以生产者消费者有三种关系，两种角色，一个交易场所。

三种关系：

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(); //队列popWakeup(&_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;
}

`

消费者慢消费，生产者快生产的执行结果：

生产者慢生产，消费者快消费的运行结果：

我们会发现，任务井然有序的执行。生产者放了数据后通知消费拿，消费者把数据拿完又会通知生产者放。