thinkphp 大型网站开发,优客工场 网站开发,金融网站框架模板下载,wordpress 移至回收站生产者消费者模型 文章目录 生产者消费者模型概念原则优点 基于BlockingQueue的生产者消费者模型BlockingQueue模拟实现单生产者消费者模型基于计算任务和存储任务的生产者消费者模型 概念
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题生产者和消费者彼…生产者消费者模型 文章目录 生产者消费者模型概念原则优点 基于BlockingQueue的生产者消费者模型BlockingQueue模拟实现单生产者消费者模型基于计算任务和存储任务的生产者消费者模型 概念
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题生产者和消费者彼此之间不直接通讯而是通过阻塞队列来进行通讯所以生产者生产完数据之后不用等待消费者处理直接扔给阻塞队列消费者不找生产者要数据而是直接从阻塞队列里取阻塞队列就相当于一个缓冲区平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的
原则 实际上生产者消费者模型本质上是维护321原则 生产者消费者模型是多线程互斥与同步的经典场景通常有以下原则 三种关系生产者与生产者互斥关系、消费者与消费者互斥关系、生产者与消费者互斥和同步关系两种角色生产者和消费者通常由线程或进程充当一个交易场所通常指的是一个特定的缓冲区临时保存数据的场所 在生产者之间或消费者之间阻塞队列作为公共资源要被多个线程访问那么就要被互斥量保护起来即作为临界资源。而互斥锁需要被多个线程竞争式申请对于生产者而言一次只允许一个生产者访问临界资源即生产者之间具有互斥关系对于消费者而言一次只允许一个消费者访问临界资源即消费者之间具有互斥关系若生产者一直生产直到缓冲区满了则生产失败而消费者一直消费直到缓冲区为空则消费失败。一边一直占用锁导致另一边的饥饿问题这是非常低效的。因此在生产者和消费者之间阻塞队列也需要被互斥量保护起来即是临界资源生产者和消费者不能同时访问即二者具有互斥关系而生产者和消费者需要具有一定的顺序访问即具有同步关系 优点 解耦支持并发支持忙先不均 我们在主函数调用目标函数实际上是强耦合主函数将参数传参给目标函数主函数传递了数据作为生产者形成变量即变量暂时保存了数据目标函数将该变量进行操作并返回即目标函数消费了数据作为消费者而主函数需要等待接收目标函数的返回值才能往下执行因此主函数和目标函数是强耦合关系。而生产者消费者模型中生产者向缓冲区中生产数据若缓冲区没满则可以一直生产消费者可以一直从缓冲区里取数据若缓冲区不为空则可以一直消费那么生产者和消费者就可以并发执行即生产者消费者模型是对强耦合关系的解耦。
基于BlockingQueue的生产者消费者模型
BlockingQueue 在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。 其与普通的队列区别在于当队列为空时从队列获取元素的操作将会被阻塞直到队列中被放入了元素当队列满时往队列里存放元素的操作也会被阻塞直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的线程在对阻塞队列进程操作时会被阻塞)
模拟实现单生产者消费者模型
为了方便理解下面以单生产者消费者为例子 main.cc #includeiostream
#includepthread.h
#includeassert.h
#includequeue
#includestdlib.h
#includectime
#includesys/types.h
#includeunistd.h
#includeblockqueue.hpp
using namespace std;void* producter(void* args)
{blockqueueint * _pbqstatic_castblockqueueint*(args);int num0;while(true){numrand()%2001;//设置随机数据_pbq-push(num);//生产者放入数据coutproducter push num: num in bqendl;sleep(1);}return nullptr;
}void* consumer(void* args)
{blockqueueint * _cbqstatic_castblockqueueint*(args);int ret0;while(true){ _cbq-pop(ret);//消费者取出数据coutconsumer take num: ret from bqendl;// sleep(1);}return nullptr;
}int main()
{srand((unsigned long)time(nullptr)^getpid());blockqueueint* _bq new blockqueueint();
pthread_t p,c;//int npthread_create(p,nullptr,producter,(void*)_bq);//生产者线程assert(n0); int mpthread_create(p,nullptr,consumer,(void*)_bq);//消费者线程assert(m0);pthread_join(p,nullptr);//回收生产者线程pthread_join(c,nullptr);//回收消费者线程delete _bq;//删除队列return 0;
}blockqueue.hpp(以.hpp开头的文件可以将定义和实现放一起调用者只需要调用该文件即可) #includeiostream
#includepthread.h
#includeassert.h
#includequeue
using namespace std;
#define GMAXCAP 5//宏定义阻塞队列的最大容量templateclass T
class blockqueue
{public:blockqueue()//构造:_maxcap(GMAXCAP){pthread_mutex_init(_mut,nullptr);//初始化互斥锁pthread_cond_init(_pcond,nullptr);//初始化生产者的条件变量pthread_cond_init(_ccond,nullptr);//初始化消费者的条件变量}void push(Tin)//输入型参数用{pthread_mutex_lock(_mut);//加锁while(is_full()){pthread_cond_wait(_pcond,_mut);//若队列为满生产者需要阻塞等待直到队列不为满才能放数据}
//走到这就放数据
_q.push(in);//往队列中放数据
pthread_mutex_unlock(_mut);//解锁
pthread_cond_signal(_ccond);//唤醒消费者线程}void pop(T* out)//输出型参数用*。输入输出型参数用{
pthread_mutex_lock(_mut);//加锁
while(is_empty())//队列为空消费者就需要阻塞等待直到队列中至少存在一个数据{pthread_cond_wait(_ccond,_mut);}
//走到这可以取数据
*out_q.front();
_q.pop();
pthread_mutex_unlock(_mut);//解锁
pthread_cond_signal(_pcond);//唤醒生产者线程}~blockqueue()
{
pthread_mutex_destroy(_mut);//释放互斥锁
pthread_cond_destroy(_pcond);//释放生产者条件变量
pthread_cond_destroy(_ccond);//释放消费者生产变量
}
bool is_full()
{return _q.size()_maxcap;//判断队列是否为满
}
bool is_empty()
{return _q.size()0;//判断队列是否为空
}private:
queueT _q;//队列
int _maxcap;//队列中的最大容量
pthread_mutex_t _mut;//互斥锁
pthread_cond_t _pcond;//生产者的条件变量
pthread_cond_t _ccond;//消费者的条件变量
};阻塞队列要给生产者往里push数据让消费者从中pop数据那么该阻塞队列需要被这两个线程所看到。因此在创建生产者线程和消费者线程时需要将阻塞队列作为参数传参给者两个线程生产者负责生产随机数并往阻塞队列中pushpush成功并打印日志消费者负责从阻塞队列中拿取数据拿取成功并打印日志我们实现的是单生产者单消费者模型需要维护生产者和消费者之间的互斥和同步关系blockqueue队列存储数据的上限为5当队列中存储了5组数据后生产者将会阻塞不能生产生产者线程在判断is_full的时候用的是while而if理由如下消费者线程相同 无论是生产者线程还是消费者线程都是先加锁再进行判断是否满足条件。以生产者线程来说若判断is_full是用的if那么有可能函数pthread_wait调用失败那么继续往后走就会出问题其次是在多生产者的情况下是可能唤醒生产者线程的函数是pthread_cond_broadcast那么情况是唤醒全部的生产者线程其实待唤醒的线程就一个就会导致伪唤醒而造成多个线程进入临界区的情况。为了避免以上情况我们又需要先加锁再判断满足条件就需要用到while来判断 当生产者push一个数据后就意味着阻塞队列中至少有一个数据若此时消费者线程在is_empty里面阻塞等待的话生产者线程就会唤醒消费者线程醒来相同的当消费者线程pop一个数据后意味着阻塞队列中至少有一个空间消费者线程就会唤醒生产者线程醒来。 生产者生产地慢消费者消费地块 生产者每生产一个数据往队列中push消费者就从队列里拿一个数据并pop呈现出生产者线程和消费者线程交错执行 生产者生产地块消费者消费地慢 可以看到先是生产者生产了五个数据并push进阻塞队列此时阻塞队列满了生产者生产失败需要消费者去pop数据然后消费者消费一个数据队列中有了空位生产者生产一个数据并push然后呈现出这两个线程交替执行
基于计算任务和存储任务的生产者消费者模型 根据图可以看到我们维护的还是单线程阻塞队列比之前的模型多了一个阻塞队列和线程 blockqueue.hpp #pragma once
#includeiostream
#includepthread.h
#includeassert.h
#includequeue
using namespace std;#define GMAXCAP 5//宏定义阻塞队列的最大容量
templateclass T
class blockqueue
{public:blockqueue()//构造:_maxcap(GMAXCAP){pthread_mutex_init(_mut,nullptr);//初始化互斥锁pthread_cond_init(_pcond,nullptr);//初始化生产者的条件变量pthread_cond_init(_ccond,nullptr);//初始化消费者的条件变量}void push(Tin)//输入型参数用{pthread_mutex_lock(_mut);//加锁while(is_full()){pthread_cond_wait(_pcond,_mut);//若队列为满生产者需要阻塞等待直到队列不为满才能放数据}
//走到这就放数据
_q.push(in);//往队列中放数据
pthread_mutex_unlock(_mut);//解锁
pthread_cond_signal(_ccond);//唤醒消费者线程}void pop(T* out)//输出型参数用*。输入输出型参数用
{
pthread_mutex_lock(_mut);//加锁
while(is_empty())//队列为空消费者就需要阻塞等待直到队列中至少存在一个数据
{pthread_cond_wait(_ccond,_mut);
}
//走到这可以取数据
*out_q.front();
_q.pop();
pthread_mutex_unlock(_mut);//解锁
pthread_cond_signal(_pcond);//唤醒生产者线程
}~blockqueue()
{
pthread_mutex_destroy(_mut);//释放互斥锁
pthread_cond_destroy(_pcond);//释放生产者条件变量
pthread_cond_destroy(_ccond);//释放消费者生产变量
}
bool is_full()
{return _q.size()_maxcap;//判断队列是否为满
}bool is_empty()
{return _q.size()0;//判断队列是否为空
}private:
queueT _q;//队列
int _maxcap;//队列中的最大容量
pthread_mutex_t _mut;//互斥锁
pthread_cond_t _pcond;//生产者的条件变量
pthread_cond_t _ccond;//消费者的条件变量
};可以看到这里的阻塞队列的.hpp文件和之前的单生产者单消费者模型的文件是一样的 main.cc #includeiostream
#includepthread.h
#includeassert.h
#includequeue
#includestdlib.h
#includectime
#includesys/types.h
#includeunistd.h
#includeblockqueue.hpp
#includetask.hpp
using namespace std;int mymath(int x,int y,char op)//计算任务需要调用的函数即计算四则运算并返回结果
{int ret0;switch(op){case :retxy;break;case -:retx-y;break; case *:retx*y;break; case /:if(y0){cerrdiv zero erro!endl;ret-1;}else{retx/y;}break; case %:if(y0){cerrdiv zero erro!endl;ret-1;}else{retx%y;}break;default://do nothingbreak;}return ret;
}templateclass C,class S//C-calculate,S-save
class blockqueues
{public:
blockqueueC* c_bq;//计算队列
blockqueueS* s_bq;//存储队列
};const string Goper-*/%;//运算符集合void* producter(void* args)//生产者
{blockqueueCaltask * _cbq(static_castblockqueuesCaltask,SaveTask*(args))-c_bq;while(true){int xrand()%2001;//设置随机数据int yrand()%100;//设置随机数据int numrand()%Goper.size();//随机运算符Caltask cal(x,y,Goper[num],mymath);//创建计算任务对象_cbq-push(cal);//生产者放入数据coutproducter push num: cal.taskstringforP() in bqendl;sleep(1);}return nullptr;
}void* consumer(void* args)//消费者
{blockqueueCaltask * _cbq(static_castblockqueuesCaltask,SaveTask*(args))-c_bq;//取出传过来的参数中的c_bq队列blockqueueSaveTask* _sbq(static_castblockqueuesCaltask,SaveTask*(args))-s_bq;//取出传过来的参数中的s_bq队列while(true){Caltask ret;_cbq-pop(ret);//消费者取出数据coutconsumer take savetask: ret() from bqendl;string resultret();SaveTask sat(result,tosave);_sbq-push(sat);coutconsumer push savetask to bqendl;// sleep(1);}return nullptr;
}void* saver(void* args)
{
blockqueueSaveTask* _sbq(static_castblockqueuesCaltask,SaveTask*(args))-s_bq;
while(true)
{
SaveTask ret;
_sbq-pop(ret);//saver取出数据
ret();//调用仿函数
coutsaver finish taskendl;
}
return nullptr;
}int main()
{srand((unsigned long)time(nullptr)^getpid());
blockqueuesCaltask,SaveTask _bqs;
_bqs.c_bqnew blockqueueCaltask();//创建计算队列
_bqs.s_bqnew blockqueueSaveTask();//创建存储队列pthread_t p,c,s;int npthread_create(p,nullptr,producter,_bqs);//生产者线程assert(n0); int mpthread_create(c,nullptr,consumer,_bqs);//消费者线程assert(m0);int kpthread_create(s,nullptr,saver,_bqs);//存储线程assert(k0);pthread_join(p,nullptr);//回收生产者线程pthread_join(c,nullptr);//回收消费者线程pthread_join(s,nullptr);//回收消费者线程delete _bqs.c_bq;//删除队列delete _bqs.s_bq;//删除队列return 0;
}创建一个blockqueues对象里面有两个成员分别是计算任务对应的队列c_bq和存储任务对应的队列s_bqdelete对象时不能直接删除blockqueues对象这样会造成内存泄漏需要删除里面的成员即队列生产者负责生产随机参数一、随机参数二、随机运算符将这些参数传递给Caltask对象然后将Caltask放入到阻塞队列c_bq中消费者负责从队列中取出Caltask对象并调用对象的仿函数进行运算并返回运算式字符串加上打印运算式然后将运算式字符串放入s_bq中并且打印日志saver负责将运算式字符串从s_bq中取出然后将运算式字符串存储到当前路径的log.txt文件中并且打印日志还需注意的是这个模型的节奏是按照生产者来的生产者隔一秒生产一个计算任务然后消费者消费一次saver存储一次即生产者生产的慢消费者消费的快从而saver存储的快 task.hpp #pragma once
#includestdlib.h
#includefunctional
using namespace std;
class Caltask
{typedef functionint(int,int,char) fun_c;
public:
Caltask(){}//无参构造Caltask(int x,int y,char op,fun_c func)
:_x(x)
,_y(y)
,_op(op)
,_caltask(func)
{}string operator()()//()运算符重载
{
int ret_caltask(_x,_y,_op);//调用外部传进来的计算任务
char buffer[128];
snprintf(buffer,sizeof buffer,%d %c %d %d,_x,_op,_y,ret);
return buffer;
}string taskstringforP()//供外部调用打印运算式字符串
{char buffer[128];snprintf(buffer,sizeof buffer,%d %c %d ?,_x,_op,_y);return buffer;
}private:
int _x;//参数一
int _y;//参数二
char _op;//运算符号
fun_c _caltask;//需调用的外部计算函数
};class SaveTask
{typedef functionvoid(string) fun_c;
public:
SaveTask(){}//默认构造SaveTask( string s,fun_c func)
:_str(s)
,_func(func)
{}void operator()()
{_func(_str);
}private:
string _str;
fun_c _func;
};void tosave(const strings)
{string target./log.txt;//文件的路径FILE*fpfopen(target.c_str(),a);// 以追加的方式打开文件if(!fp)//文件打开失败{coutfopen errorendl;}fputs(s.c_str(),fp);//往文件里面写数据fputs(\n,fp);// 往文件里面写换行符fclose(fp);//关闭文件
}
