基于blockqueue的生产和消费模型

线程篇下讲的是基于阻塞队列的生产者消费者模型。在学习这个之前我们先了解一些其他概念：

同步：在保证数据安全的条件下，让线程按某种特定的顺序依次访问临界资源。

通过上一节的代码我们实现了一个多线程抢票的程序，但结果显示的是一个线程在疯狂的抢票，这就导致了其他线程抢占不到临界资源而导致的饥饿问题。

所以在抢代码的代码逻辑中，我们需要保证各个线程以同步的方式进行抢票。那如何实现呢？就需要用到信号量：

条件变量的接口函数：

条件变量的使用：

#include <iostream>
#include <string>
#include <unistd.h>
#include <pthread.h>int tickets = 1000;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;void *start_routine(void *args)
{std::string name = static_cast<const char *>(args);while (true){pthread_mutex_lock(&mutex);pthread_cond_wait(&cond, &mutex); // 为什么要有mutex，后面就说// 判断暂时省略std::cout << name << " -> " << tickets << std::endl;tickets--;pthread_mutex_unlock(&mutex);}
}int main()
{// 通过条件变量控制线程的执行pthread_t t[5];for (int i = 0; i < 5; i++){char *name = new char[64];snprintf(name, 64, "thread %d", i + 1);pthread_create(t+i, nullptr, start_routine, name);}while (true){sleep(1);// pthread_cond_signal(&cond);pthread_cond_broadcast(&cond);std::cout << "main thread wakeup one thread..." << std::endl;}for (int i = 0; i < 5; i++){pthread_join(t[i], nullptr);}return 0;
}

条件变量的理解：

我们用一个例子来说明：

没有条件变量就好比是面试官在一个办公室面试，只要应聘者得到门外挂着的钥匙进来就可以面试。这时一个应聘者抢到了钥匙进入了办公室，面试结束后想要将钥匙归还时感觉自己面的不好，这时自己离钥匙也是最近的，所以刚出门又拿起了钥匙继续面试。这钥匙就导致了其他应聘者拿不到而这个应聘者在疯狂的面试，很明显这不合理。

但是有了条件变量就好比是面试官在外面设置了一个等待区，面试完的人必须归还钥匙并且再次回到等待队列的最末尾重新排队，这就保证了各个应聘者都有机会面试。

 条件不满足的时候，线程必须去某些定义好的条件变量上进行等待。

pthread_cond_wait第二个参数的意义（将锁传进去）:该函数调用的时候，会以原子性的方式，将锁释放，并将线程挂起等待。用pthread_cond_sign将线程唤醒时，该线程会拿着你所传入的锁将代码继续向下运行。通过以上知识的铺垫让我们进入今天的正题：

生产者消费者模型

什么是生产者消费者模型呢？我们用一个形象的图来说明：

生产者消费者模型就是生产者可以将生产的数据存放在一个共享区，消费者从共享区中获得数据进行消费。由于是共享区，就要保证数据的安全性。当生产者生产一个数据时，另一个生产者不能在同一个数据上生产，不然会导致数据的不安全性。因此生产者和生产者之间是互斥关系的。同样消费者也不能同时消费同一个数据，不然可能会导致数据的不一致性，因此消费者和消费者之间是互斥关系的。如果这时一个生产者正在生产一个数据，而同时消费者也正在消费这个数据。就可能导致数据还没生产完全就已经被消费了，也会导致数据的不安全性。因此生产者和消费者之间是互斥关系，同时我们还希望生产者生产以后就有消费者来消费，消费者消费完一个就有生产者来生产，因此生产者和消费者之间是同步关系。

总结一下就是3种关系、两种角色、一个交易场所：

这一段特殊的缓冲区可以提前存放一批数据，这样消费者想消费的时候消费，生产者想什么时候生产就什么时候生产。解决了生产和消费两批线程忙闲不均的问题，是它们不具有强耦合的关系。

基于blockqueue的生产和消费模型

基于普通方式处理数据

代码实现：

//Main.cc#include <iostream>
#include <unistd.h>
#include <time.h>
#include "BlockQueue.hpp"using namespace std;void* consumer(void* args)
{BlockQueue<int>* bq =static_cast<BlockQueue<int>*>(args);while(true){int val =0;bq->pop(&val);cout<<"我是消费者，我消费了一个数字："<<val<<endl;sleep(1);}return nullptr;
}void* productor(void* args)
{BlockQueue<int>* bq =static_cast<BlockQueue<int>*>(args);while(true){int val =rand()%10;bq->push(val);cout<<"我是生产者,我生产了一个数字："<<val<<endl;//sleep(1);}return nullptr;
}int main()
{srand((unsigned long)time(nullptr)^getpid());BlockQueue<int>* queue =new BlockQueue<int>(5);pthread_t c,p;pthread_create(&c,nullptr,consumer,(void*)queue);pthread_create(&p,nullptr,productor,(void*)queue);pthread_join(c,nullptr);pthread_join(p,nullptr);return 0;
}//BlockQueue.hpp
#include <queue>
#include <pthread.h>
#include <iostream>
using namespace std;const int NUM =5;
template<class T>
class BlockQueue
{
public:BlockQueue(const int numsize =NUM):_maxsize(NUM){pthread_mutex_init(&_lock,nullptr);pthread_cond_init(&_ccond,nullptr);pthread_cond_init(&_ccond,nullptr);}void push(const T& in){pthread_mutex_lock(&_lock);while(is_Full()){//这里说明阻塞队列是满的，需要让生产者等待pthread_cond_wait(&_pcond,&_lock);}//这里说明阻塞队列至少有一个空位可以插入_queue.push(in);//唤醒消费者去消费pthread_cond_signal(&_ccond);pthread_mutex_unlock(&_lock);}void pop(T* out){pthread_mutex_lock(&_lock);while(is_Empty()){//这里说明阻塞队列是空的，需要让消费者等待pthread_cond_wait(&_ccond,&_lock);}//这里说明阻塞队列至少有一个数据*out=_queue.front();_queue.pop();//唤醒生产者生产数据pthread_cond_signal(&_pcond);pthread_mutex_unlock(&_lock);}~BlockQueue(){pthread_mutex_destroy(&_lock);pthread_cond_destroy(&_ccond);pthread_cond_destroy(&_pcond);}private:bool is_Full(){return _queue.size()==_maxsize;}bool is_Empty(){return _queue.empty();}private:queue<T> _queue;int _maxsize;pthread_mutex_t _lock;   //保护临界资源的锁pthread_cond_t _ccond;   //消费者的条件变量pthread_cond_t _pcond;   //生产者的条件变量
};

代码细节：

基于计算器任务的Task

我们得创建一个task.hpp,里面定义一个CalTask类：

class CalTask
{
public:using func_t =std::function<int(int,int,char)>;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[64];snprintf(buffer,sizeof buffer,"%d %c %d =%d",_x,_op,_y,result);return buffer;}std::string to_string(){char buffer[64];snprintf(buffer,sizeof buffer,"%d %c %d = ?",_x,_op,_y);return buffer;}~CalTask(){}private:int _x;int _y;char _op;func_t _callback;};

生产者任务：

void* productor(void* args)
{BlockQueue<CalTask>* bq =static_cast<BlockQueue<CalTask>*>(args);while(true){int x =rand()%10;int y =rand()%10;char op =op_str[rand()%op_str.size()];CalTask task(x,y,op,mymath);bq->push(task);cout<<"productor task:"<<task.to_string()<<endl;//sleep(1);}return nullptr;
}

消费者任务：

void* consumer(void* args)
{BlockQueue<CalTask>* bq =static_cast<BlockQueue<CalTask>*>(args);while(true){CalTask t;bq->pop(&t);string result =t();cout<<"consumer result:"<<result<<endl;sleep(1);}return nullptr;
}

基于两个阻塞队列实现计算与存储：

将计算和存储的两个队列放进同一个类中：

template<class c,class s>
class BlockQueues
{public:BlockQueue<c>* c_bq; BlockQueue<s>* s_bq;
};

main函数：

int main()
{srand((unsigned long)time(nullptr)^getpid());BlockQueues<CalTask,SaveTask> bqs;bqs.c_bq =new BlockQueue<CalTask>;bqs.s_bq =new BlockQueue<SaveTask>;//BlockQueue<CalTask>* bqs =new BlockQueue<CalTask>(5);pthread_t c,p,s;pthread_create(&c,nullptr,consumer,(void*)&bqs);pthread_create(&p,nullptr,productor,(void*)&bqs);pthread_create(&s,nullptr,saver,(void*)&bqs);pthread_join(c,nullptr);pthread_join(p,nullptr);pthread_join(s,nullptr);delete bqs.c_bq;delete bqs.s_bq;return 0;
}

存储任务：

class SaveTask
{typedef std::function<void(const std::string&)> func_t;
public:SaveTask(){}SaveTask(const std::string &message, func_t func): _message(message), _func(func){}void operator()(){_func(_message);}
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){std::cerr << "fopen error" << std::endl;return;}fputs(message.c_str(), fp);fputs("\n", fp);fclose(fp);
}

存储线程执行的任务：

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();std::cout << "save thread,保存任务完成..." << std::endl; }return nullptr;
}

代码的执行结果：

生产者与消费者模型优势总结：

这个模型的优势不在于多个线程能够并发式的对共享资源里面的数据进行访问和处理，而是多个线程能够在加载任务和处理任务的时候进行并发处理。

在我们上面的代码逻辑中很简单，就是随机构造x和y构成一个任务最后放入阻塞队列中。而在实际情况中加载任务的时候没有那么简单，有时需要从网络或者数据库中加载，这就需要消耗很长的一段时间。这个模型的优势就是在于多个生产者能同时加载多个任务，随后竞争出一名生产者将任务放入共享资源（阻塞队列）中，然后在竞争出一名消费者取出任务。因此模型的优势不在于多个线程能并发的从阻塞队列中拿数据处理，而是在加载任务的时候做到并发节省时间。同理处理任务，当一个线程处理任务的同时，另一个线程仍可以从阻塞队列取出任务处理，不影响之前的进程处理任务，因此消费者处理任务的环节也做到了并发，实现了加载任务和处理任务的解耦，提高了整个程序的运行效率和速度。

到这里本章的内容就全部结束了，创作不易，希望大家多多点赞支持。