生产者消费者模型是一种常见的并发模式，用于解决生产者和消费者之间的协作问题。它是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，而不用去关心消费者是否要取的问题；消费者不找生产者要数据，而是直接从阻塞队列里取，而不关心生产者是否在生产的问题。

所以阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力，这个阻塞队列就是用来给生产者和消费者解耦的！

但是因为生产者和消费者是异步的，所以必须使用同步机制来保证它们之间的正确协作，避免数据竞争和死锁等问题。

举个例子，生产者就是工厂，消费者就是我们日常消费的人，一般来说，我们是不会直接去工厂买东西的，而是到超市去买，超市就相对于是一个缓冲区，我们每次买东西的时候都会去超市买，超市一般都会储备比较多的货，所以消费者一般不需要去关心是否有货的问题；而工厂每次生产完货之后就像超市投放，并且可以等到超市缺货的时候再去生产，就有了一段空白时间可以干其它的事情，并且也不需要去关心是否有人去消费的问题，起到了相互解耦的作用！

Ⅱ. 生产者消费者模型的优点

解耦合：生产者和消费者之间的通信是通过缓冲区进行的，彼此之间无需直接通信，从而实现了解耦合，提高了系统的可维护性和可扩展性。
提高效率：生产者可以一次性生产多个产品，然后将它们批量存储在缓冲区中，消费者可以一次性消费多个产品，这样可以大大提高生产和消费的效率。
减少资源浪费：如果没有缓冲区，生产者必须阻塞等待消费者消费产品后才能生产下一个产品，而消费者也必须阻塞等待生产者生产产品后才能消费产品，这样会浪费大量的资源，降低了系统的效率。
支持并发：生产者消费者模型是多线程编程中的经典案例，可以帮助程序员更好地理解和掌握多线程编程的技巧和方法。

Ⅲ. 基于阻塞队列的生产者消费者模型

基于 阻塞队列（Blocking Queue） 的生产者消费者模型是一种常见的实现方式，其与普通的队列区别在于：（以下的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞）

队列为空，从阻塞队列获取元素的操作将会被阻塞，直到阻塞队列中被放入了资源。
队列为满，往阻塞队列里存放元素的操作也会被阻塞，直到阻塞队列中有空闲位置。

下面我们写一个程序，实现功能：一个就是我们的生产者可以通过第一个阻塞队列向队列中放入任务（可以自定义），然后消费者向队列中拿任务并且执行。除此之外，当消费者向第一个队列中拿出任务并且得到结果的时候，同时向第二个阻塞队列中放入该执行结果，充当第二个阻塞队列中的生产者。而第二个阻塞队列中的消费者则拿到这个生产的任务，实现将其生产者的执行结果写入文本文件中，达到存储的目的！

下面的程序中我们主要以四则运算为执行任务来测试：

Makefile

mythread : test.cpp
	g++ -o $@ $^ -std=c++11 -lpthread
 
.PHONY : clean
clean:
	rm -f mythread log.txt

Block_queue.hpp

#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
 
static const int MAXCAP = 500;
 
// 阻塞队列
template <class T>
class BlockQueue
{
public:
    BlockQueue(const int& maxcap = MAXCAP)
        :_maxcap(maxcap)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_pcond, nullptr);
        pthread_cond_init(&_ccond, nullptr);
    }
    
    // 生产操作
    void put(const T& in) // 输入型参数：const&
    {
        pthread_mutex_lock(&_mutex);
 
        // 这里必须使用while，而不能是if
        while(is_full()) // 若队列为满则不能放数据
        {
            pthread_cond_wait(&_pcond, &_mutex);
        }
        // 走到这说明队列不为满，可以放资源
        _bq.push(in);
        pthread_cond_signal(&_ccond); // 唤醒消费者线程
        pthread_mutex_unlock(&_mutex);
    }
    
    // 消费操作
    void take(T* out) // 输出型参数：*   // 输入输出型参数：&
    {
        pthread_mutex_lock(&_mutex);
 
        // 这里必须使用while，而不能是if
        while(is_empty()) // 若队列为空不能取资源
        {
            pthread_cond_wait(&_ccond, &_mutex);
        }
        // 走到这说明队列不为空，可以取资源
        *out = _bq.front(); 
        _bq.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() const
    {
        return _bq.size() == 0;
    }
 
    bool is_full() const
    {
        return _bq.size() == _maxcap;
    }
private:
    std::queue<T> _bq;
    int _maxcap; // 队列中元素的上限
    pthread_mutex_t _mutex; // 互斥锁
    pthread_cond_t _pcond; // 生产者对应的条件变量
    pthread_cond_t _ccond; // 消费者对应的条件变量
};

task.hpp

#pragma once
#include <iostream>
#include <functional>
#include <pthread.h>
#include <unistd.h>
#include <cstdio>
#include <string>
#include <fstream>
// 计算任务
class CalTask
{
    using func_t = std::function<int(int, int, char)>;
public:
    CalTask()
    {}
    CalTask(int x, int y, char op, func_t callback)
        :_x(x), _y(y), _op(op), _callback(callback)
    {}
    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 caltask(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 << "除零错误" << std::endl;
            result = -1;
        }
        else
        {
            result = x / y;
        }
    }
        break;
    default:
        break;
    }
    return result;
}
 
// 保存任务
class SaveTask
{
    using func_t = std::function<void(const std::string&)>;
public:
    SaveTask()
    {}
    SaveTask(const std::string &message, func_t callback)
        : _message(message), _callback(callback)
    {}
    void operator()()
    {
        _callback(_message);
    }
private:
    std::string _message;
    func_t _callback;
};
 
void savetask(const std::string& message)
{
    // c++的文件读写方式
    std::fstream fs("./log.txt", std::fstream::out | std::fstream::app);
    fs << message << "\n";
    fs.close();
}

test.cpp

#include "Block_queue.hpp"
#include "task.hpp"
#include <sys/types.h>
#include <stdlib.h>
#include <unistd.h>
#include <time.h>
#include <string>
 
// 封装一下两个阻塞队列
template <class C, class S> // C:计算  S:存储
class BlockQueues
{
public:
    BlockQueues()
    {
        cal_bq = new BlockQueue<CalTask>();
        save_bq = new BlockQueue<SaveTask>();
    }
    ~BlockQueues()
    {
        delete cal_bq;
        delete save_bq;
    }
public:
    BlockQueue<C> *cal_bq;
    BlockQueue<S> *save_bq;
};
 
// 生产者线程
void* Productor(void* bqs)
{
    BlockQueue<CalTask>* bq = (static_cast<BlockQueues<CalTask, SaveTask>*>(bqs))->cal_bq;
    while(true)
    {
        // 生产任务
        int x = rand() % 100 + 1;
        int y = rand() % 100;
        char op = oper[rand() % oper.size()];
 
        CalTask ct(x, y, op, caltask);
        bq->put(ct);
        std::cout << "productor thread, 生产计算任务: " << ct.toTaskString() << std::endl;
 
        sleep(1);
    }
    return nullptr;
}
 
// 消费者线程
void* Consumer(void* bqs)
{
    BlockQueue<CalTask>* bq = (static_cast<BlockQueues<CalTask, SaveTask>*>(bqs))->cal_bq;
    BlockQueue<SaveTask>* save_bq = (static_cast<BlockQueues<CalTask, SaveTask>*>(bqs))->save_bq;
    while(true)
    {
        // 消费任务
        CalTask t;
        bq->take(&t);
 
        // 执行任务
        std::string result = t(); // 任务非常耗时！！
        std::cout << "Consumer thread，完成计算任务: " << result << " ... done"<< std::endl;
 
        // 生产任务
        SaveTask st(result, savetask);
        save_bq->put(st);
        std::cout << "Consumer thread，推送存储任务完成..." << std::endl; 
        sleep(1);
    }
    return nullptr;
}
 
// 保存者/消费者线程
void* Saver(void* bqs)
{
    BlockQueue<SaveTask>* save_bq = (static_cast<BlockQueues<CalTask, SaveTask>*>(bqs))->save_bq;
    while(true)
    {
        // 消费任务
        SaveTask st;
        save_bq->take(&st);
 
        // 执行任务
        std::cout << "save thread，保存任务完成..." << std::endl; 
        st();
    }
    return nullptr;
}
 
int main()
{
    srand((unsigned int)time(nullptr) ^ getpid()); // 生成随机种子
 
    BlockQueues<CalTask, SaveTask> bqs;
    pthread_t consumer, productor, saver;
    pthread_create(&productor, nullptr, Productor, &bqs);
    pthread_create(&consumer, nullptr, Consumer, &bqs);
    pthread_create(&saver, nullptr, Saver, &bqs);
 
    pthread_join(consumer, nullptr);
    pthread_join(productor, nullptr);
    pthread_join(saver, nullptr);
    return 0;
}

上面只是三个单独的生产者消费者线程在执行，其实我们还可以产生更多的生产者或者消费者同时执行，读者自行尝试！

Ⅳ. 如何理解提高了效率❓❓❓

生产者消费者模型可以提高效率的原因在于它可以有效地利用多核处理器的并行性，从而充分利用系统资源。例如，在一个多线程的程序中，生产者可以在一个线程中运行，而消费者可以在另一个线程中运行，这样就可以同时进行数据的生产和处理，而不必等待一个操作完成后再执行另一个操作。

要知道，其实这种模型是串行的方式通过阻塞队列的，所以其实相对于生产者直接向消费者投递资源、消费者直接向生产者拿资源来说其实效率不是高的很明显，而关键提高效率的要点在于生产者投放资源是需要消耗时间的，这段时间对于其它的生产者来说是不影响的可以继续投放，提高了效率；对于消费者来说，拿取资源的时候，也是需要消耗时间的，但是这段时间内其它的消费者也能拿取资源。总的来说，也就是阻塞队列是在不断工作的，效率真正高在生产者和消费者生产资源和消费资源所消耗的这段时间，并不影响阻塞队列中其它生产者和消费者的工作！

此外，生产者消费者模型还可以通过使用缓冲区来平衡生产者和消费者之间的速度差异。当生产者生产数据的速度快于消费者处理数据的速度时，数据可以暂存在缓冲区中，以免数据丢失或者降低系统的效率。反之，当消费者处理数据的速度快于生产者生产数据的速度时，缓冲区中可以存储更多的数据，以便消费者随时获取数据。

总之，生产者消费者模型通过合理的分配任务、优化资源利用、平衡生产和消费速度等手段，可以提高系统的效率。

Ⅰ. 生产者消费者模型的概念

文章目录