基于BlockQueue的生产消费模型及Linux中的信号量

基于BlockQueue的生产消费模型

Task.hpp

#pragma once#include<cstdio>
#include<iostream>
#include<string>
#include<functional>using namespace std;
class CalTask
{using func_t=function<int(int,int,char)>;//typedef function<int(int,int)> func_t;
public:CalTask(){}CalTask(int x,int y,char op,func_t func):_x(x),_y(y),_op(op),_callback(func){}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;}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;
};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){cerr<<"div zero error!!!"<<endl;result=-1;}elseresult=x/y;}break;case '%':{if(y==0){cerr<<"mod zero error!!!"<<endl;result=-1;}elseresult=x%y;}          break;default:break;}return result;
}
class SaveTask
{typedef function<void(const string&)> func_t;
public:SaveTask(){}SaveTask(const string &message,func_t func):_message(message),_func(func){}void operator()(){_func(_message);}
private:string _message;func_t _func;
};
void Save(const string& message)
{const string target="./log.txt";FILE* fp=fopen(target.c_str(),"a+");if(!fp){cerr<<" fopen error "<<endl;return;} fputs(message.c_str(),fp);fputs("\n",fp);fclose(fp);
}

Block.hpp

#pragma once#include<iostream>
#include<queue>
#include<pthread.h>
#include<unistd.h>using namespace std;
static const int gmaxcap=5;
template<class T>
class BlockQueue
{
public:BlockQueue(const int& maxcap=gmaxcap):_maxcap(maxcap){pthread_mutex_init(&_mutex,nullptr);pthread_cond_init(&_pcond,nullptr);pthread_cond_init(&_ccond,nullptr);}void push(const T& in)//输入型参数，const &{pthread_mutex_lock(&_mutex);//1.判断while(is_full())//防止向队列中多push内容->提高代码健壮性{//生产条件不满足无法生产，需要生产者等待//临界区://pthread_cond_wait   第二个参数必须为我们正在使用的互斥锁！！！//1.pthread_cond_wait:该函数在调用的时候，会以原子性的方式将锁释放，并将自己挂起//2.pthread_cond_wait:该函数在被唤醒返回的时候，会自动重新获取当时传入的锁pthread_cond_wait(&_pcond,&_mutex);}_q.push(in);//pthread_cond_signal即可在临界区内也可在临界区外pthread_cond_signal(&_ccond);pthread_mutex_unlock(&_mutex);}void pop(T* out)//输出型参数：*， // 输入输出型：&{pthread_mutex_lock(&_mutex);while(is_empty()){pthread_cond_wait(&_ccond,&_mutex);}*out=_q.front();_q.pop();pthread_cond_signal(&_pcond);pthread_mutex_unlock(&_mutex);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_pcond);pthread_cond_destroy(&_ccond);}
private:bool is_empty() {return _q.empty();}bool is_full(){return _q.size()==_maxcap;}queue<T> _q;int _maxcap;//队列中元素的上限pthread_mutex_t _mutex;pthread_cond_t _pcond;//生产者对应的条件变量pthread_cond_t _ccond;//消费者对应的条件变量
};

Main.cc

#include<time.h>
#include<cstring>
#include<sys/types.h>
#include"BlockQueue.hpp"
#include"Task.hpp"const string oper="+-*/%";template<class C,class S>
class BlockQueues
{
public:BlockQueue<C> *c_bq;BlockQueue<S> *s_bq;
};
void* productor(void *bqs_)
{//BlockQueue<Task>* bq=static_cast<BlockQueue<Task>*>(bq_);BlockQueue<CalTask>* bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->c_bq;while(true){//生产活动-x先用随机数构建数据int x=rand()%10+1;int y=rand()%5;int operCode=rand()%oper.size();CalTask t(x,y,oper[operCode],mymath);bq->push(t);cout<<"productor thread,生产数据: "<<t.toTaskString()<<endl;sleep(1);}return nullptr;
}
void *consumer(void* bqs_)
{//BlockQueue<Task>* bq=static_cast<BlockQueue<Task>*>(bq_);BlockQueue<CalTask>* bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->c_bq;BlockQueue<SaveTask>* save_bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->s_bq;while(true){//消费活动CalTask t;bq->pop(&t);string result=t();cout<<"消费任务: "<<result<<endl;// SaveTask save(result,Save);// save_bq->push(save);// cout<<"consumer thread,推送保存任务完成..."<<endl;}return nullptr;
}void* saver(void* bqs_)
{BlockQueue<SaveTask>* save_bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->s_bq;while(true){SaveTask t;save_bq->pop(&t);t();cout<<"保存任务完成..."<<endl;}
}
int main()
{srand((unsigned long)time(nullptr));BlockQueues<CalTask,SaveTask> bqs;bqs.c_bq=new BlockQueue<CalTask>();bqs.s_bq=new BlockQueue<SaveTask>();pthread_t c,p,s;pthread_create(&c,nullptr,consumer,&bqs);pthread_create(&p,nullptr,productor,&bqs);// pthread_create(&s,nullptr,saver,&bqs);pthread_join(c,nullptr);pthread_join(p,nullptr);// pthread_join(s,nullptr);delete bqs.c_bq;delete bqs.s_bq;return 0;
}

POSIX信号量

POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。但POSIX可以用于线程间同步。

        pthread_mutex_lock(&_mutex);while(is_empty()){pthread_cond_wait(&_ccond,&_mutex);}*out=_q.front();_q.pop();pthread_cond_signal(&_pcond);pthread_mutex_unlock(&_mutex);