40 生产者消费者模型

生产者消费者模型

概念

为何要使用生产者消费者模型，这个是用过一个容器解决生产者和消费的强耦合问题。生产者和消费者之间不需要通讯，通过阻塞队列通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列取，阻塞队列相当于缓冲区，平衡了双方的能力，用来解耦的

上面超市的例子。消费者需要泡面的话不用去找供货商要货，而是去超市取。如果找供货商，消费者只需要一包，供货商开启生产设备只生产一包，多次这样很浪费效率也不高。超市作为存储，需要一万包泡面，供货商生产1万包摆到超市里，将超市塞满，缓存起来，调整供货商和消费者的速度不一致导致的效率问题。供货商就可以休息下来。消费者需要几包去超市取，支持了一种忙闲不均的状态。供货商关注超市有多少空位置，需要多少货，消费者关注现有商品的数量。供货商在生产的时候，和消费者没关系，消费者购买的时候和供应商也没关系，双方不需要互相考虑，只完成自己的事情，就减少了依赖性，解耦。

在计算机里，生产者和消费者都是由线程承担，超市是一种特殊结构的内存空间，这个结构是一种共享资源，整个过程就是执行流在通信，如何安全高效的通信。共享资源就有并发的问题，这种并发有三种关系：

生产者和生产者：互斥关系。一个在供货的时候另一个需要等待
生产者和消费者：互斥和同步关系。如果供货商正在摆一个商品，消费者有没有得到。有一种不确定性，生产者要确定，数据安全，只有生产了和没有生产，消费者一定可以得到货物。供货商不停联系超市需不需要货，超市已经满了还在不停询问，占用了消费者询问的机会，导致消费者饥饿问题。所以要同步，保证顺序性。供货商刚供货一次再询问时，告知一段时间之内不要询问，消费者询问没有商品时，也告知没有并一段时间不要询问，安全才是本质
消费者和消费者：互斥关系

321原则
3种关系2种角色1个交易场所
3种关系，生产者和消费者之间互相搭配
2种角色：生产和消费
1个交易场所：特定结构的内存空间

优点：
1.支持忙闲不均
2.支持并发
3.生产和消费进行解耦

什么是解耦。main函数内调用一个函数，传入参数，需要等到函数返回才能继续往下执行，可以将两个分开为线程，参数用一段空间缓存，放到缓冲区里，函数调用时在空间里取，这就是解耦

基于BlockingQueue的生产者模型

BlockingQueue

在多线程编程中阻塞队列（Blocking Queue）是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出（以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞）
类似于管道

类的设计
首先需要数据存储的结构，这个用队列，一个容量设置为队列允许存放的最多数量。对队列的访问同一时间只能有一方，所以需要一个锁。类提供存入数据和取出数据的功能，生产者关心的是还能放多少数据，如果大于最大容量就要停止，所以要判断队列的现有数量，这是对共享资源的访问，加锁和释放锁，判断大于容量时就去条件变量队列等待，同样，消费者取物品也需要一个条件变量，消费者判断有没有商品，没有就到消费者的条件变量等待。当生产者生产出一个商品放入后就唤醒消费者取，消费者取完唤醒生产者生产

#include <queue>
#include <pthread.h>template<class T>
class BlockQueue
{static const int defaultnum = 20;
public:BlockQueue(int cap = defaultnum){_maxcap = cap;pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_pcond, nullptr);pthread_cond_init(&_ccond, nullptr);_lowwater = _maxcap / 3;_highwater = _maxcap / 3 * 2;}void push(const T& x){pthread_mutex_lock(&_mutex);if (_que.size() == _maxcap) //防止被伪唤醒的状态 {pthread_cond_wait(&_pcond, &_mutex); //调用,自动释放锁}_que.push(x);  //确保生产条件满足才能生产//if (_que.size() > _highwater)pthread_cond_signal(&_ccond);pthread_mutex_unlock(&_mutex);}T pop(){pthread_mutex_lock(&_mutex);if (_que.size() == 0){pthread_cond_wait(&_ccond, &_mutex);}T tmp = _que.front();_que.pop();//if (_que.size() < _lowwater)pthread_cond_signal(&_pcond);pthread_mutex_unlock(&_mutex);return tmp;}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_pcond);pthread_cond_destroy(&_ccond);}private:std::queue<T> _que;int _maxcap;   //最大容量pthread_mutex_t _mutex;pthread_cond_t _pcond;pthread_cond_t _ccond;int _lowwater;int _highwater; //控制水位线
};

阻塞队列设置为了模板，不只可以放入内置类型，也可以是自定义类型。弄一个任务类，有两个操作数，操作符加减乘除随机。一个变量记录结果，一个记录可靠性，如果有除0错误设置为对应值。提供返回string类型整个表达式的内容功能

#pragma once
#include <stdio.h>
#include <string>enum
{DIVZERO = 1,UNKNOW
};
std::string g_op = "+-*/";
struct task
{
public:task(int a, int b, char op):_a(a), _b(b), _op(op), _result(0), _exitcode(0){}void run(){switch(_op){case '+':_result = _a + _b;break;case '-':_result = _a - _b;break;case '*':_result = _a * _b;break;case '/':if (_b == 0){_exitcode = DIVZERO;}else{_result = _a / _b;}      break;default:_exitcode = UNKNOW;break;}//printf("%d+%d结果:%d\n", _a, _b, _a + _b);}std::string getresult(){std::string str = std::to_string(_a) + _op  + std::to_string(_b);str += "=";str += std::to_string(_result);str += " [exit:";str += std::to_string(_exitcode);str += "]";return str;}std::string gettask(){std::string str = std::to_string(_a) + _op  + std::to_string(_b);return str;}private:int _a;int _b;char _op;int _result;int _exitcode;
};

main文件生成两个线程，生产和消费，传入阻塞队列的实例，生产出一个任务，消费者完成

#include <unistd.h>
#include <cstdlib>
#include <iostream>
#include <ctime>
#include "blockqueue.hpp"
#include "task.hpp"void *produce(void *bk)
{BlockQueue<task>* block = static_cast<BlockQueue<task>*>(bk);while (true){int x1 = rand() % 10;usleep(10);int x2 = rand() % 10 + 1;char op = g_op[rand() % 4];task t(x1, x2, op);//生产printf("生产任务:%s\n", t.gettask().c_str());block->push(t);sleep(1);}
}void* consume(void* bk)
{BlockQueue<task>* block = static_cast<BlockQueue<task>*>(bk);while (true){//消费task n = block->pop();n.run();printf("完成任务:%s\n", n.getresult().c_str());sleep(1);}
}int main()
{srand(time(NULL));pthread_t ptid, ctid;BlockQueue<task>* block = new BlockQueue<task>();pthread_create(&ptid, nullptr, produce, block);pthread_create(&ctid, nullptr, consume, block);while (true){sleep(1);}delete block;return 0;
}

结果：

伪唤醒

当队列里只剩一个位置的时候，生产者如果不小心唤醒了多个生产者。这时它们都会去竞争锁，拿到锁的线程去生产然后放入，接着释放锁。正常情况下，应该消费线程拿到这个锁取数据，但因为刚刚唤醒了多个线程，可能会抢到锁继续放入数据，这时就会超出最大容量出现错误。所以要将if处改为循环，释放锁后判断如果满了就调条件变量里休眠

多生产多消费

将上面的单生成单消费改为多生产多消费版本

#include <unistd.h>
#include <cstdlib>
#include <iostream>
#include <ctime>
#include "blockqueue.hpp"
#include "task.hpp"void *produce(void *bk)
{BlockQueue<task>* block = static_cast<BlockQueue<task>*>(bk);while (true){int x1 = rand() % 10;usleep(10);int x2 = rand() % 10 + 1;char op = g_op[rand() % 4];task t(x1, x2, op);//生产printf("%p生产任务:%s\n", pthread_self(), t.gettask().c_str());block->push(t);sleep(1);}
}void* consume(void* bk)
{BlockQueue<task>* block = static_cast<BlockQueue<task>*>(bk);while (true){//消费task n = block->pop();n.run();printf("%p完成任务:%s\n", pthread_self(), n.getresult().c_str());//sleep(1);}
}int main()
{srand(time(NULL));BlockQueue<task>* block = new BlockQueue<task>();pthread_t ptid[3], ctid[5];for (int i = 0; i < 3; i++){pthread_create(&ptid[i], nullptr, produce, block);}for (int i = 0; i < 5; i++){pthread_create(&ctid[i], nullptr, consume, block);}for (int i = 0; i < 3; i++){pthread_join(ptid[i], nullptr);}for (int i = 0; i < 5; i++){pthread_join(ctid[i], nullptr);} delete block;return 0;
}

优势

虽然同一时间只能有一个执行流访问阻塞队列，多个生产者也只能有一个访问队列，那多个生产者和消费者的优势体现在什么地方。
生产者的数据从用户网络等地方获得，数据的获取也需要时间，当一个生产者往队列里放入数据时，其他生产者可以同时获取数据，后面只需要放入数据即可。消费者方拿到数据后要对数据加工处理，这部分也是需要花费时间，同样一个线程获取数据时，其他的可能正在处理获得的数据。所以说，这个模型提高了效率，并发程度，是高效的。