【Linux】25. 生产者消费者模型

生产者消费者模型

321原则(便于记忆)

为何要使用生产者消费者模型

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

生产者消费者模型优点

• 解耦
• 支持并发
• 支持忙闲不均

a

基于BlockingQueue的生产者消费者模型

BlockingQueue

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

C++ queue模拟阻塞队列的生产消费模型

阻塞队列的生产消费模型代码实现

// BlockQueue.hpp代码逻辑
#pragma once

#include <iostream>
#include <queue>
#include <pthread.h>

const int gmaxcap = 4;

template <class T>
class BlockQueue
{
public:
    // 构造函数
    BlockQueue(const int &maxcap = gmaxcap) : _maxcap(maxcap)
    {
        // 将锁和条件变量全部初始化
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_pcond, nullptr);
        pthread_cond_init(&_ccond, nullptr);
    }

    void push(const T &in) // 输入型参数 const & ???
    {
        // 加锁 -- 确保成功
        pthread_mutex_lock(&_mutex);
        // 1. 判断
        // 用while是关键点！
        // 如果用if判断 可能发生假唤醒(直接往后执行)
        // 用while后 线程每次唤醒后都会重新检查判断条件是否满足
        while(is_full())
        {
            // 细节：pthread_cond_wait函数的第二个参数必须是我们正在使用的互斥锁！
            // a. pthread_cond_wait : 该函数调用时会以原子性的方式，将锁释放并将自己挂起
            // b. pthread_cond_wait : 该函数被唤醒返回时，会自动重新获取传入的锁！！！(函数底层实现)
            // 不会出现挂起锁被拿走 导致程序无法运行
            // 也不会出现唤醒没锁拿的情况

            // 队列中满了 生产条件不满足，无法生产，生产者进行等待
            pthread_cond_wait(&_pcond,&_mutex); 
        }

        // 2. 线程走到这里一定是没有满
        _q.push(in);

        // 3. 阻塞队列当中一定存在数据
        // 细节: pthread_cond_signal：这个函数，可以放在临界区内部，也可以放在外部
        pthread_cond_signal(&_ccond); //唤醒消费线程-让消费线程进行消费
        
        pthread_mutex_unlock(&_mutex); // 解锁
    }
    void pop(T *out) // 输出型参数：*
    {
        pthread_mutex_lock(&_mutex); // 加锁
        // 1. 判断
        while(is_empty()) // 用while理由同上
        {
            // 队列空了，无法进行消费 让消费线程阻塞等待
            pthread_cond_wait(&_ccond,&_mutex);
        }

        // 线程走到这里确保队列一定不为空
        *out = _q.front();
        _q.pop(); // 取出数据

        // 取出1个数据就确保队列当中至少空出来了一个位置！
        // 唤醒生产线程
        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()
    {
        return _q.empty();
    }
    bool is_full()
    {
        return _q.size() == _maxcap;
    }
private:
    std::queue<T> _q;
    int _maxcap;            // 队列当中元素的上限
    pthread_mutex_t _mutex; // 锁
    pthread_cond_t _pcond;  // 生产者对应的条件变量
    pthread_cond_t _ccond;  // 消费者对应的条件变量
    // 为啥两个条件变量呢 -- 生产者/消费者都需要控制(排队)
};

// BlockQueue.cc 的代码逻辑

#include "BlockQueue.hpp"
#include "Task.hpp"
#include <sys/types.h>
#include <unistd.h>
#include <ctime>

// 想要实现的功能： 生产者生产数据 消费者消费数据
// 生产者随机生成数字以及运算符 消费者得出结果并返回

// C:计算
// S:存储

template <class C, class S>
class BlockQueues
{
public:
    BlockQueue<C> *c_bq;
    BlockQueue<S> *s_bq;
};
// 生产数据(生产者)
void *productor(void *bqs_)
{
    BlockQueue<CalTask> *bq = (static_cast<BlockQueues<CalTask, SaveTask> *>(bqs_))->c_bq;

    while (true)
    {
        sleep(5);
        int x = rand() % 100 + 1; // 用随机数构建数据
        int y = rand() % 10;
        int operCode = rand() % oper.size();
        // 传入的参数在Task.hpp当中都有定义
        CalTask t(x, y, oper[operCode], mymath);

        // 将任务插入生产队列
        bq->push(t);
        std::cout << "productor thread,生产计算任务: " << t.toTaskString() << std::endl;
    }
    return nullptr;
}
// 计算数据并提交数据 (消费者+生产者)
void *consumer(void *bqs_)
{
    BlockQueue<CalTask> *bq = (static_cast<BlockQueues<CalTask, SaveTask> *>(bqs_))->c_bq;
    BlockQueue<SaveTask> *save_bq = (static_cast<BlockQueues<CalTask, SaveTask> *>(bqs_))->s_bq;

    while(true)
    {
        // 消费活动
        CalTask t;
        bq->pop(&t);

        std::string result = t();
        std::cout << "cal thread,完成计算任务: " << result << "... done" << std::endl;
    
        SaveTask save(result,Save);
        save_bq->push(save);
        std::cout << "cal thread,推送保存任务完成..." << std::endl;

        sleep(1);
    }
    return nullptr;
}
// 保存数据(消费者)
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;
}

int main()
{
    // srand 随机数种子 通篇只需要设置1次 ^ getpid 伪随机 确保每次的随机性
    srand((unsigned long)time(nullptr) ^ getpid());

    BlockQueues<CalTask, SaveTask> bqs;
    // 创建计算和保存队列
    bqs.c_bq = new BlockQueue<CalTask>();
    bqs.s_bq = new BlockQueue<SaveTask>();

    // 创建2个生产者，2个消费者线程
    pthread_t c[2], p[2], s;
    pthread_create(p, nullptr, productor, &bqs);
    pthread_create(p + 1, nullptr, productor, &bqs);
    // pthread_create(p + 2, nullptr, productor, &bqs);
    sleep(1);
    pthread_create(c, nullptr, consumer, &bqs);
    pthread_create(c + 1, nullptr, consumer, &bqs);

    // 创建1个保存线程
    pthread_create(&s, nullptr, saver, &bqs);

    pthread_join(c[0], nullptr);
    pthread_join(c[1], nullptr);
    pthread_join(p[0], nullptr);
    pthread_join(p[1], nullptr);
    pthread_join(p[2], nullptr);
    pthread_join(s, nullptr);

    delete bqs.c_bq;
    delete bqs.s_bq;
    return 0;
}

生产消费模型的高效体现(重点)

POSIX信号量

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()

上一节生产者-消费者的例子是基于queue的,其空间可以动态分配,现在基于固定大小的环形队列重写这个程序（POSIX信号量）

基于环形队列的生产消费模型

• 环形队列采用数组模拟，用模运算来模拟环状特性
  
  环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来
  判断满或者空。另外也可以预留一个空的位置，作为满的状态

但是我们现在有信号量这个计数器，就很简单的进行多线程间的同步过程

环形队列的生产消费模型代码实现

// ring_queue.hpp 代码逻辑

#pragma once

#include <iostream>
#include <vector>
#include <cassert>
#include <semaphore.h>
#include <pthread.h>

static const int gcap = 4;

// 环形队列 -- 用模板后续可以传入任务
template <class T>
class RingQueue
{
private:
    // 信号量需要保证原子性 P/V操作
    void P(sem_t &sem)
    {
        int n = sem_wait(&sem); // 信号量--
        assert(n == 0);
        (void)n;
    }
    void V(sem_t &sem)
    {
        int n = sem_post(&sem); // 信号量++
        assert(n == 0);
        (void)n;
    }
    // sem_wait和sem_post 申请不到的时候将线程阻塞等待直到能申请再唤醒

public:
    // 队列容量空间为cap 容量大小为cap
    RingQueue(const int &cap = gcap) : _queue(cap), _cap(cap)
    {
        // 初始化信号量
        int n = sem_init(&_spaceSem, 0, _cap); // 空间资源一开始为cap
        assert(n == 0);
        n = sem_init(&_dataSem, 0, 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::vector<T> _queue;
    int _cap; // 容量大小

    // 定义两个信号量
    sem_t _spaceSem; // 生产者看重的是空间资源
    sem_t _dataSem;  // 消费者看重的是数据资源

    int _productorStep;
    int _consumerStep;

    pthread_mutex_t _pmutex; // 生产者锁
    pthread_mutex_t _cmutex; // 消费者锁
};

// Task.hpp 逻辑 之前实现的模块 进行简化

// 确保头文件只被包含一次
// 为了避免由于多次包含同一个头文件而导致的重复定义错误
#pragma once

#include <iostream>
#include <string>
#include <cstdio>
#include <functional>

// 任务
class Task
{
    using func_t = std::function<int(int, int, char)>;

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;
        }
        else
            result = x / y;
    }
    break;
    case '%':
    {
        if (y == 0)
        {
            std::cerr << "mod zero error!" << std::endl;
            result = -1;
        }
        else
            result = x % y;
    }
    break;
    default:
        // do nothing
        break;
    }
    return result;
}

// main.cc 代码逻辑

#include "ring_queue.hpp"
#include "Task.hpp"
#include <pthread.h>
#include <ctime>
#include <cstdlib>
#include <sys/types.h>
#include <unistd.h>

std::string SelfName()
{
    char name[128];
    snprintf(name, sizeof(name), "thread[0x%x]", pthread_self());
    return name;
}

void *ProductorPoutine(void *rq)
{
    RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
    while (true)
    {
        sleep(2);
        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;
    }
}

void *ConsumerRoutine(void *rq)
{
    RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
    while (true)
    {
        // sleep(2);
        Task t;
        // 消费任务
        ringqueue->Pop(&t);
        std::string result = t();
        std::cout << SelfName() << ",消费者消费了一个任务:" << result << std::endl;
    }
}

int main()
{
    // 随机数种子
    srand((unsigned int)time(nullptr) ^ getpid() ^ pthread_self()); // 足够随机
    RingQueue<Task> *rq = new RingQueue<Task>();                    // 构建生产者-消费者队列

    pthread_t p[2], c[2];
    for (int i = 0; i < 2; i++)
        pthread_create(p + i, nullptr, ProductorPoutine, rq);
    for (int i = 0; i < 2; i++)
        pthread_create(c + i, nullptr, ConsumerRoutine, rq);

    for (int i = 0; i < 2; i++)
        pthread_join(p[i], nullptr);
    for (int i = 0; i < 2; i++)
        pthread_join(c[i], nullptr);

    delete rq;
    return 0;
}

同理，生产者慢就按照生产者的节奏走(生产一个消费一个)