基于blockqueue的生产和消费模型

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

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

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

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

条件变量的接口函数：

条件变量的使用：

#include 
#include 
#include 
#include 
 
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 
#include 
#include 
#include "BlockQueue.hpp"
 
 
using namespace std;
 
 
void* consumer(void* args)
{
    BlockQueue<int>* bq =static_castint>*>(args);
    while(true)
    {
        int val =0;
        bq->pop(&val);
        cout<<"我是消费者，我消费了一个数字："<
        sleep(1);
    }
 
    return nullptr;
}
 
 
void* productor(void* args)
{
    BlockQueue<int>* bq =static_castint>*>(args);
    while(true)
    {
        int val =rand()%10;
        bq->push(val);
        cout<<"我是生产者,我生产了一个数字："<
        //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 
#include 
#include 
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 _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* bq =static_cast*>(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:"<to_string()<
        //sleep(1);
    }
    return nullptr;
}

消费者任务：

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

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

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

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

int main()
{
    srand((unsigned long)time(nullptr)^getpid());
    BlockQueues bqs;
    
  
    bqs.c_bq =new BlockQueue;
    bqs.s_bq =new BlockQueue;
    //BlockQueue* bqs =new BlockQueue(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 *save_bq = (static_cast *>(bqs_))->s_bq;
 
    while(true)
    {
        SaveTask t;
        save_bq->pop(&t);
        t();
        std::cout << "save thread,保存任务完成..." << std::endl; 
    }
    return nullptr;
}