网站域名怎么做解析,网站建设 唐山,搜索量查询百度指数,wordpress 安装中文字体文章目录 一、生产者消费者模型二、基于BlockingQueue的生产者消费者模型1.BlockQueue.hpp2.Task.hpp3.main.cc 三、POSIX信号量四、基于环形队列的生产消费模型1.RingQueue.hpp2.Task.hpp3.main.cc 一、生产者消费者模型 我们这里举一个例子#xff0c;来解释生产者消费者模… 文章目录 一、生产者消费者模型二、基于BlockingQueue的生产者消费者模型1.BlockQueue.hpp2.Task.hpp3.main.cc 三、POSIX信号量四、基于环形队列的生产消费模型1.RingQueue.hpp2.Task.hpp3.main.cc 一、生产者消费者模型 我们这里举一个例子来解释生产者消费者模型我们学生–消费者供应商–生产者超市–交易场所我们买东西只需要关系售货架子上是否有商品即可没有了商品超市从供应商进行供货。供应商和供应商不能同时向一个货架进行供货所以生产者之间是互斥的关系非消费者和消费不能同时从同一个货架拿商品所以消费者与消费者之间是互斥的关系而消费者需要等生产者将商品放到货架之后才能拿取商品所以生产者和消费者之间是互斥和同步的关系。 生产消费模型: 生产者和生产者之间:互斥关系 消费者和消费者之间:互斥关系 生产者和消费者之间:互斥同步 总结:“321”原则: 3种关系:生产者和生产者互斥)消费者和消费者互斥)生产者和消费者 互斥保证共享资源的安全性]同步)–产品数据) 种角色:生产者线程消费者线程 1个交易场所:(一段特定结构的缓冲区 只要我们想写生产消费模型我们本质工作其实就是维护321原则! 挖掘特点: 1.生产线程和消费线程进行解耦 2支持生产和消费的一段时间的忙闲不均的问题 3.提高效率 生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯而通过阻塞队列来进行通讯所以生产者生产完数据之后不用等待消费者处理直接扔给阻塞队列消费者不找生产者要数据而是直接从阻塞队列里取阻塞队列就相当于一个缓冲区平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。 生产者消费者模型优点 解耦 支持并发 支持忙闲不均 二、基于BlockingQueue的生产者消费者模型 BlockingQueue 在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于当队列为空时从队列获取元素的操作将会被阻塞直到队列中被放入了元素当队列满时往队列里存放元素的操作也会被阻塞直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的线程在对阻塞队列进程操作时会被阻塞) 1.BlockQueue.hpp #pragma once#include iostream #include queue #include pthread.husing namespace std; const int gnum 5;template class T class BlockQueue { public:BlockQueue(const int maxcap gnum): _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);// 1. 判断// 细节2: 充当条件判断的语法必须是while不能用ifwhile (is_full()){// 细节1pthread_cond_wait这个函数的第二个参数必须是我们正在使用的互斥锁// a. pthread_cond_wait: 该函数调用的时候会以原子性的方式将锁释放并将自己挂起// b. pthread_cond_wait: 该函数在被唤醒返回的时候会自动的重新获取你传入的锁pthread_cond_wait(_pcond, _mutex);}// 2. 走到这里一定是没有满_q.push(in);// 3. 绝对能保证阻塞队列里面一定有数据// 细节3pthread_cond_signal这个函数可以放在临界区内部也可以放在外部pthread_cond_signal(_ccond);pthread_mutex_unlock(_mutex);}void pop(T *out){pthread_mutex_lock(_mutex);// 1. 判断while (is_empty()){pthread_cond_wait(_ccond, _mutex);}// 2. 走到这里我们能保证一定不为空*out _q.front();_q.pop();// 3. 绝对能保证阻塞队列里面至少有一个空的位置pthread_cond_signal(_pcond);pthread_mutex_unlock(_mutex);}~BlockQueue(){pthread_mutex_destroy(_mutex);pthread_cond_destroy(_pcond);pthread_cond_destroy(_ccond);}private:bool is_full(){return _q.size() _maxcap;}bool is_empty(){return _q.empty();}private:queueT _q;int _maxcap; // 队列中元素的上限pthread_mutex_t _mutex;pthread_cond_t _pcond; // 生产者对应的条件变量pthread_cond_t _ccond; // 消费者对应的条件变量 }; 2.Task.hpp #pragma once#include iostream #include string #include functionalclass CalTask { public:typedef std::functionint(int, int, char) func_t;// using func_t std::functionint(int, int, char);public:CalTask(){}CalTask(int x, int y, char op, func_t func): _x(x), _y(y), _op(op), _callback(func){}std::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;}std::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 std::string oper -*/%;int calculate(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){std::cerr div zero error std::endl;return -1;}elseresult x / y;}break;case %:{if (y 0){std::cerr mod zero error std::endl;return -1;}elseresult x % y;}break;default:std::cerr 请输入正确的操作符 std::endl;break;}return result; }class SaveTask { public:using func_t functionvoid(const std::string );public:SaveTask(){}SaveTask(const std::string message, func_t func): _message(message), _func(func){}void operator()(){_func(_message);}~SaveTask(){}private:std::string _message;func_t _func; };void Save(const std::string message) {const std::string target ./log.txt;FILE *fp fopen(target.c_str(), a);if (fp nullptr){perror(fopen error);return;}fputs(message.c_str(), fp);fputs(\n, fp);fclose(fp); }3.main.cc #include BlockQueue.hpp #include Task.hpp#include ctime #include unistd.h #include sys/types.h// C计算 // S: 存储 template class C, class S class BlockQueues { public:BlockQueueC *_cbq;BlockQueueS *_sbq; };void *productor(void *args) {BlockQueueCalTask *bq static_castBlockQueuesCalTask, SaveTask *(args)-_cbq;while (true){int x rand() % 10 1;int y rand() % 5;char op oper[rand() % 5];CalTask t(x, y, op, calculate);bq-push(t);std::cout productor thread 生产计算任务: t.toTaskString() std::endl;sleep(1);}return nullptr; }void *consumer(void *args) {BlockQueueCalTask *bq static_castBlockQueuesCalTask, SaveTask *(args)-_cbq;BlockQueueSaveTask *save_bq static_castBlockQueuesCalTask, SaveTask *(args)-_sbq;while (true){CalTask t;bq-pop(t);std::string result t();std::cout cal thread,完成计算任务: result ... done std::endl;SaveTask st(result, Save);save_bq-push(st);cout cal thread,推送存储任务完成... std::endl;sleep(1);}return nullptr; }void *saver(void *args) {BlockQueueSaveTask *save_bq static_castBlockQueuesCalTask, SaveTask *(args)-_sbq;while (true){SaveTask st;save_bq-pop(st);st();std::cout save thread,保存任务完成... std::endl;// sleep(1);}return nullptr; }int main() {BlockQueuesCalTask, SaveTask bqs;bqs._cbq new BlockQueueCalTask();bqs._sbq new BlockQueueSaveTask();pthread_t p, c, s;pthread_create(p, nullptr, productor, bqs);pthread_create(c, nullptr, consumer, bqs);pthread_create(s, nullptr, saver, bqs);pthread_join(p, nullptr);pthread_join(c, nullptr);pthread_join(s, nullptr);delete bqs._cbq;delete bqs._sbq;return 0; }你创建多线程生产和消费的意义是什么??2.生产消费模型高效在哪里?? 可以在生产之前和消费之后让线程并行执行 生产者而言向blockqueue里面放置任务 他的任务从哪里来的呢?它获取任务和构建任务要不要花时间 消费者而言从blockqueue里面拿取任务 对于消费者难道他把任务从任务队列中拿出来就完了吗??消费者拿到任务之后后续还有没有任务?? 三、POSIX信号量 先发现我们之前写的代码的不足的地方 pthread_mutex_lock(_mutex); while (is_full()) {pthread_cond_wait(_pcond, _mutex); } _q.push(in); pthread_cond_signal(_ccond); pthread_mutex_unlock(_mutex);1.一个线程在操作临界资源的时候必须临界资源是满足条件的! 2.可是公共资源是否满足生产或者消费条件我们无法、直接得知【我们不能事前得知【在没有访问之前无法得知】】 3.只能先加锁再检测再操作再解锁。因为你要检测本质:也是在访问临界资源! 因为我们在操作临界资源的时候有可能不就绪但是我们无法提前得知所以只能先加锁在检测根据检测结果决定下一步怎么走! 只要我们对资源进行整体加锁就默认了我们对这个资源整体使用。实际情况可能存在:.一份公共资源但是允许同时访问不同的区域!程序员编码保证不同的线程可以并发访问公共资源的不同区域! 什么是信号量: a.信号量本质是一把计数器。衡量临界资源中资源数量多少的计数器 b.只要拥有信号量就在未来一定能够拥有临界资源的一部分申请信号量的本质:对临界资源中特定小块资源的预订机制 线程要访问临界资源中的某一区域–申请信号量–所有人必须的先看到信号量–信号量本身必须是:公共资源 有可能我们在访问真正的临界资源之前我们其实就可以提前知道临界资源的使用情况! ! ! 信号量只要申请成功就一定有你的资源。只要申请失败就说明条件不就绪你只能等!!不需要在判断了! 计数器–递减or 递增sem_t sem 10; sem–;—申请资源—必须保证操作的原子性— P sem;—归还资源—必须保证操作的原子性----V 信号量核心操作:PV原语 POSIX信号量和SystemV信号量作用相同都是用于同步操作达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步。 初始化信号量 #include semaphore.h int sem_init(sem_t *sem, int pshared, unsigned int value); 参数 pshared:0表示线程间共享非零表示进程间共享 value信号量初始值销毁信号量 int sem_destroy(sem_t *sem);等待信号量 功能等待信号量会将信号量的值减1 int sem_wait(sem_t *sem); //P()发布信号量 功能发布信号量表示资源使用完毕可以归还资源了。将信号量值加1。 int sem_post(sem_t *sem);//V()四、基于环形队列的生产消费模型 环形队列采用数组模拟用模运算来模拟环状特性 环形结构起始状态和结束状态都是一样的不好判断为空或者为满所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置作为满的状态 但是我们现在有信号量这个计数器就很简单的进行多线程间的同步过程 生产和消费在什么情况下可能访问同一个位置: 1.空的时候 2.满的时候 3.其他情况生产者和消费者根本访问的就是不同的区域! 为了完成环形队列cp问题我们要做的核心工作是什么 1.你不能超过我 2我不能把你套一个圈以上 3我们两个什么时候会站在一起? 信号量是用来衡量临界资源中资源数量的 1.对于生产者而言看中什么?队列中的剩余空间—空间资源定义十个信号量 2.对于消费者而言看中的是什么?放入队列中的数据! —数据资源定义一个信号量 生产者而言: prodocter_sem: 0 // 申请成功你就可以继续向下运行。 //申请失败当前执行流阻塞在申请处 P(producter_sem); //从事生产活动--把数据放入到队列中 V(comsumer_sem);消费者而言: comsumer_sem:10 P(comsumer_sem); //从事消费活动 v(producter_sem);_未来生产和消费的位置我们要想清楚: 1.其实就是队列中的下标、 2一定是两个下标 3.为空或者为满下标相同 1.RingQueue.hpp #pragma once#include iostream #include string #include vector #include cassert #include semaphore.hstatic const int gcap 5;template class T class RingQueue { private:// 等待信号量会将信号量的值减1void P(sem_t sem){int n sem_wait(sem);assert(n 0);(void)n;}// 发布信号量表示资源使用完毕可以归还资源了。将信号量值加1。void V(sem_t sem){int n sem_post(sem);assert(n 0);(void)n;}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);pthread_mutex_lock(_pmutex);_queue[_productorStep] in;_productorStep % _cap;pthread_mutex_unlock(_pmutex);V(_dataSem);}// 消费者void pop(T *out){P(_dataSem);pthread_mutex_lock(_cmutex);*out _queue[_consumerStep];_consumerStep % _cap;pthread_mutex_unlock(_cmutex);V(_spaceSem);}~RingQueue(){sem_destroy(_spaceSem);sem_destroy(_dataSem);pthread_mutex_destroy(_pmutex);pthread_mutex_destroy(_cmutex);}private:std::vectorT _queue;int _cap;sem_t _spaceSem; // 生产者 想生产看中的是什么资源呢? 空间资源sem_t _dataSem; // 消费者 想消费看中的是什么资源呢? 数据资源int _productorStep;int _consumerStep;pthread_mutex_t _pmutex;pthread_mutex_t _cmutex; };2.Task.hpp #pragma once#include iostream #include string #include cstdio #include functionalclass Task {using func_t std::functionint(int,int,char);// typedef std::functionint(int,int) func_t; public:Task(){}Task(int x, int y, char op, func_t func):_x(x), _y(y), _op(op), _callback(func){}std::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;}std::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 std::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){std::cerr div zero error! std::endl;result -1;}elseresult x / y;}break;case %:{if (y 0){std::cerr mod zero error! std::endl;result -1;}elseresult x % y;}break;default:// do nothingbreak;}return result; }3.main.cc #include RingQueue.hpp #include Task.hpp #include pthread.h #include ctime #include cstdlib #include sys/types.h #include unistd.hstd::string SelfName() {char name[128];snprintf(name, sizeof(name), thread[0x%x], pthread_self());return name; }void *ProductorRoutine(void *rq) {// RingQueueint *ringqueue static_castRingQueueint *(rq);RingQueueTask *ringqueue static_castRingQueueTask *(rq);while (true){// version1// int data rand() % 10 1;// ringqueue-Push(data);// std::cout 生产完成生产的数据是 data std::endl;// version2// 构建or获取任务 --- 这个是要花时间的int x rand() % 10;int y rand() % 5;char op oper[rand() % oper.size()];Task t(x, y, op, mymath);// 生产任务ringqueue-push(t);// 输出提示std::cout SelfName() , 生产者派发了一个任务: t.toTaskString() std::endl;// sleep(1);} }void *ConsumerRoutine(void *rq) {// RingQueueint *ringqueue static_castRingQueueint *(rq);RingQueueTask *ringqueue static_castRingQueueTask *(rq);while (true){// version1// int data;// ringqueue-Pop(data);// std::cout 消费完成消费的数据是 data std::endl;// sleep(1);// version2Task t;// 消费任务ringqueue-pop(t);std::string result t(); // 消费也是要花时间的std::cout SelfName() , 消费者消费了一个任务: result std::endl;// sleep(1);} }int main() {srand((unsigned int)time(nullptr) ^ getpid() ^ pthread_self() ^ 0x71727374);// RingQueueint *rq new RingQueueint();RingQueueTask *rq new RingQueueTask();// 单生产单消费多生产多消费 -- 只要保证最终进入临界区的是一个生产一个消费就行// 多生产多消费的意义pthread_t p[4], c[8];for (int i 0; i 4; i)pthread_create(p i, nullptr, ProductorRoutine, rq);for (int i 0; i 8; i)pthread_create(c i, nullptr, ConsumerRoutine, rq);for (int i 0; i 4; i)pthread_join(p[i], nullptr);for (int i 0; i 8; i)pthread_join(c[i], nullptr);delete rq;return 0; }