基于阻塞队列及环形队列的生产消费模型

ops/2024/10/10 17:51:43/

目录

条件变量函数

等待条件满足

阻塞队列

升级版

信号量

POSIX信号量

环形队列

条件变量函数

等待条件满足

int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex); 参数： cond：要在这个条件变量上等待 mutex：互斥量，后面详细解释

pthread_cond_wait：第二个参数必须是正在使用的互斥锁

a.pthread_cond_wait：该函数调用时，会以原子性的方式将锁释放，并将自己挂起

b.pthread_cond_wait：该函数被唤醒返回的时候，会自动从新获取锁

阻塞队列

blockqueue.hpp

#pragma once
#include<iostream>
#include<string>
#include<queue>
#include<unistd.h>
#include<pthread.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){pthread_mutex_lock(&_mutex);while (is_full())pthread_cond_wait(&_pcond,&_mutex);//生产条件不满足_q.push(in);//阻塞队列中一定有数据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;}
private:queue<T> _q;int _maxcap;pthread_mutex_t _mutex;pthread_cond_t _pcond;//生产者对应的条件变量pthread_cond_t _ccond;
};

task.hpp

#pragma once
#include<iostream>
#include<cstdio>
#include<string>
#include<functional>using namespace std;
class Task
{using func_t=function<int(int,int,char)>;
public:Task(){}Task(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;
};

升级版

blockqueue.hpp

#pragma once
#include<iostream>
#include<string>
#include<queue>
#include<unistd.h>
#include<pthread.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){pthread_mutex_lock(&_mutex);while (is_full())pthread_cond_wait(&_pcond,&_mutex);//生产条件不满足_q.push(in);//阻塞队列中一定有数据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;}
private:queue<T> _q;int _maxcap;pthread_mutex_t _mutex;pthread_cond_t _pcond;//生产者对应的条件变量pthread_cond_t _ccond;
};

task.hpp

#pragma once
#include<iostream>
#include<cstdio>
#include<string>
#include<functional>using namespace std;
class CalTask
{using func_t=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){}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;
};
const 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){cerr<<"div zero error!"<<endl;result=-1;}    else result=x/y;}break;case '%':{if(y==0){cerr<<"div zero error!"<<endl;result=-1;}    else result=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);
}

MainCp.cc

#include"BlockQueue.hpp"
#include"task.hpp"
#include<sys/types.h>
#include<unistd.h>
#include<ctime>//
//
template<class C,class S>
class BlockQueues
{public:BlockQueue<C>* c_bq;BlockQueue<S>* s_bq;
};
void* consumer(void* bqs_)
{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){/* consumer */// int data;// bq->pop(&data);CalTask t;bq->pop(&t);string result=t();cout<<"消费数据: "<<result<<endl;SaveTask save(result,Save);save_bq->push(save);cout<<"推送保存任务完成..."<<endl;sleep(1);}return nullptr;
}
void* producter(void* bqs_)
{BlockQueue<CalTask>* bq=(static_cast<BlockQueues<CalTask,SaveTask>* >(bqs_))->c_bq;while (true){//producerint x=rand()%10+1;int y=rand()%5;int operCode=rand()%oper.size();CalTask t(x,y,oper[operCode],mymath);bq->push(t);cout<<"生产任务: "<<t.toTaskString()<<endl;// sleep(1);}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;}return nullptr;
}
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,producter,&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;
}

./MainCp
生产任务: 9 * 0 = ? 
生产任务: 9 - 4 = ? 
生产任务: 8 - 0 = ? 
生产任务: 3 - 4 = ? 
生产任务: 6 + 1 = ? 
消费数据: 9 * 0 = 0 
推送保存任务完成...
生产任务: 2 - 2 = ? 
推送保存任务完成...
消费数据: 9 - 4 = 5 
推送保存任务完成...
生产任务: 9 - 0 = ? 
推送保存任务完成...
消费数据: 8 - 0 = 8 
推送保存任务完成...
生产任务: 6 * 3 = ? 
推送保存任务完成...
消费数据: 3 - 4 = -1 
推送保存任务完成...
生产任务: 4 * 4 = ? 
推送保存任务完成...
消费数据: 6 + 1 = 7 
推送保存任务完成...
生产任务: 5 % 4 = ? 
推送保存任务完成...
^C
zhangsan@ubuntu:~/practice-using-ubuntu/20241005/blockqueue$ cat log.txt
9 * 0 = 0 
9 - 4 = 5 
8 - 0 = 8 
3 - 4 = -1 
6 + 1 = 7 

信号量

a.信号量的本质就是计数器

b.只有拥有信号量，在未来就一定能拥有临界资源的一部分

申请信号量的本质就是：对临界资源中特点小块资源的预定机制

sem--         申请资源       P        必须保证操作的原子性

sem++       释放资源        V       必须保证操作的原子性

POSIX信号量

环形队列

RingQueue.hpp

#pragma once#include<iostream>
#include<cassert>
#include<vector>
#include<ctime>
#include<cstdlib>
#include<semaphore.h>
#include<unistd.h>
#include<pthread.h>static const int gcap=5;template<class T>
class RingQueue
{
private:void P(sem_t& sem){int n=sem_wait(&sem);assert(n==0);}void V(sem_t& sem){int n=sem_post(&sem);assert(n==0);}
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);}void Push(const T& in){P(_spaceSem);//productorpthread_mutex_lock(&_pmutex);_queue[_productorStep++]=in;_productorStep%=_cap;pthread_mutex_unlock(&_pmutex);//更高效V(_dataSem);}void Pop(T* out){pthread_mutex_lock(&_cmutex);P(_dataSem);*out=_queue[_consumerStep++];_consumerStep%=_cap;V(_spaceSem);pthread_mutex_unlock(&_cmutex);}~RingQueue(){sem_destroy(&_spaceSem);sem_destroy(&_dataSem);pthread_mutex_destroy(&_pmutex);pthread_mutex_destroy(&_cmutex);}
private:vector<T> _queue;int _cap;sem_t _spaceSem;//生产者->空间资源sem_t _dataSem;int _productorStep;int _consumerStep;pthread_mutex_t _pmutex;pthread_mutex_t _cmutex;
};

task.hpp

#pragma once
#include<iostream>
#include<cstdio>
#include<string>
#include<functional>using namespace std;
class Task
{using func_t=function<int(int,int,char)>;
public:Task(){}Task(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;
};
const 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){cerr<<"div zero error!"<<endl;result=-1;}    else result=x/y;}break;case '%':{if(y==0){cerr<<"div zero error!"<<endl;result=-1;}    else result=x%y;}break;default:break;}return result;
}

main.cc

#include"RingQueue.hpp"
#include"task.hpp"using namespace std;void* ProductorRoutine(void* rq)
{RingQueue<Task>* ringqueue=static_cast<RingQueue<Task>* >(rq);while (true){/* code */int x=rand()%100;int y=rand()%50;char op=oper[rand()%oper.size()];Task t(x,y,op,mymath);ringqueue->Push(t);cout<<"生产者派发了一个任务: "<<t.toTaskString()<<endl;sleep(1);}
}
void* ConsumerRoutine(void* rq)
{RingQueue<Task>* ringqueue=static_cast<RingQueue<Task>* >(rq);while (true){/* code */Task t;ringqueue->Pop(&t);string result=t();cout<<"消费者消费了一个任务"<<result<<endl;}}
int main()
{srand((unsigned int)time(nullptr));RingQueue<Task>* rq=new RingQueue<Task>();pthread_t p,c;pthread_create(&p,nullptr,ProductorRoutine,rq);pthread_create(&c,nullptr,ConsumerRoutine,rq);pthread_join(p,nullptr);pthread_join(c,nullptr);delete rq;return 0;
}


http://www.ppmy.cn/ops/123597.html

相关文章

C++——继承

C++——继承

目录 引言 继承的概念和定义 1.继承的概念 2.继承的定义 2.1 继承的语法形式 2.2 继承中类的叫法 2.3 继承后的子类成员访问权限 基类与派生类的赋值转换 1.派生类对象赋值给基类对象 2.派生类对象的引用赋值给基类对象 3.派生类对象的指针赋值给基类对象 4.基类指…
阅读更多...
jpa String index out of range: 0

jpa String index out of range: 0

ORM Spring Data Jpa 查询报错信息 原因&#xff1a;数据库字段设置为char类型&#xff0c;但是该字段为空字符串 参考&#xff1a;https://stackoverflow.com/questions/24380846/hibernate-sql-exception-java-lang-stringindexoutofboundsexception-string-index java.lang…
阅读更多...
CentOS 7 上安装 Kibana

CentOS 7 上安装 Kibana

以下是在 CentOS 7 上安装 Kibana 的步骤&#xff1a; 一、安装 Java&#xff08;如果尚未安装&#xff09; Kibana 需要 Java 运行环境。如果系统中没有安装 Java&#xff0c;可以使用以下命令安装 OpenJDK&#xff1a; sudo yum install java-1.8.0-openjdk二、添加 Elast…
阅读更多...
为什么numpy.array的数据像是字典一样，但是这个数据有real属性，又无法读取shape，显示0-d array

为什么numpy.array的数据像是字典一样，但是这个数据有real属性，又无法读取shape，显示0-d array

根据您提供的信息&#xff0c;您似乎在处理一个0维的NumPy数组&#xff0c;也称为标量。在NumPy中&#xff0c;0维数组是一个只有一个元素的数组&#xff0c;它没有行和列的结构&#xff0c;只包含一个值。这个值可以是任何数据类型&#xff0c;包括整数、浮点数或字符串。 当…
阅读更多...
华为---MUX VLAN简介及示例配置

华为---MUX VLAN简介及示例配置

目录 1. 产生背景 2. 应用场景 3. 主要功能 4. 基本概念 5. 配置步骤及相关命令 6.示例配置 6.1 示例场景 6.2 网络拓扑图 6.3 配置代码 6.4 配置及解析 6.5 测试验证 配置注意事项 1. 产生背景 MUX VLAN&#xff08;Multiplex VLAN&#xff09;提供了一种通过VLA…
阅读更多...
运用MinIO技术服务器实现文件上传——在Linux系统上安装和启动（一）

运用MinIO技术服务器实现文件上传——在Linux系统上安装和启动（一）

# MinIO 单机版环境搭建详解 ## 1. 简介 随着大数据时代的到来&#xff0c;数据存储的需求日益增大&#xff0c;如何有效地存储和管理大规模的非结构化数据成为许多企业和开发者面临的挑战。MinIO 作为一个高性能、分布式对象存储系统&#xff0c;致力于为用户提供简单、快速…
阅读更多...
C语言之扫雷小游戏（完整代码版）

C语言之扫雷小游戏（完整代码版）

说起扫雷游戏&#xff0c;这应该是很多人童年的回忆吧&#xff0c;中小学电脑课最常玩的必有扫雷游戏&#xff0c;那么大家知道它是如何开发出来的吗&#xff0c;扫雷游戏背后的原理是什么呢&#xff1f;今天就让我们一探究竟&#xff01; 扫雷游戏介绍 如下图&#xff0c;简…
阅读更多...
常见大模型架构模式

常见大模型架构模式

以下是几种常见的大模型架构模式&#xff1a; 1. 路由分发架构模式 工作原理 当用户输入一个Prompt查询时&#xff0c;该查询会被发送到路由转发模块。路由转发模块对输入Prompt进行分类。如果Prompt查询是可以识别的&#xff0c;那么它会被路由到小模型进行处理。小模型通常具…
阅读更多...
最新文章

Copyright @ 2022~2023