【Linux】生产者消费者模型

Myth丶恋晨 2024-04-25 20:18 85阅读 0赞

## 生产者消费者模型 ##

### 什么是生产者消费者模型 ###

生产者消费者问题（英语：Producer-consumer problem），也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了共享固定大小缓冲区的两个线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。  
.  
要解决该问题，就必须让生产者在缓冲区满时休眠（要么干脆就放弃数据），等到下次消费者消耗缓冲区中的数据的时候，生产者才能被唤醒，开始往缓冲区添加数据。同样，也可以让消费者在缓冲区空时进入休眠，等到生产者往缓冲区添加数据之后，再唤醒消费者。通常采用进程间通信的方法解决该问题。如果解决方法不够完善，则容易出现死锁的情况。出现死锁时，两个线程都会陷入休眠，等待对方唤醒自己。该问题也能被推广到多个生产者和消费者的情形。

**通过一个例子来理解生产者消费者模型**

相信大家通过上面的描述，肯定不了解生产者消费者模型到底是指的什么意思，因此我们利用下面这个例子来让大家了解

> 在日常生活中，我们作为消费者是会进场去超市里购物的，超市这个角色本身没有生产能力，我们所买的东西都是由工厂供应商提供的。而超市的角色就是把供应商生产的东西放在超市内的架子上来供消费者买。

![c42987e53ca7b5316882a596f5b6684e.png][]

上述超市的作用很明显：

*  提高效率（站在供应商角度，它一次性可以批发大量商品给超市；站在消费者角度，购买方便）。
 *  解耦（供应商可以随时随地供应，消费者可以随时随地购买，消费者在消费期间不影响工厂生产，工厂生产期间不影响消费者消费，二者之间不再是强耦合关系）

> 我们发现上述的消费者和工厂供应商都有多个，那他们之间是什么关系呢？
> 
> ❓生产者和生产者之间是什么关系呢？?竞争关系—互斥关系！
> 
> ❓消费者和消费者之间是什么关系呢？?竞争关系—互斥关系！
> 
> ❓那消费者和生产者之间是什么关系呢？
> 
> ?在此之前，我们先来了解一下互斥关系和同步关系！
> 
>  *  互斥关系：假设我跟小明都去超市买可乐（我们两个都只想买可乐，其他的不想买），此时超市只剩下一瓶可乐，因此我们两个人都要竞争这最后一瓶可乐，我买了小明就不能买，小明买了我就不能买，因此我们是互斥关系！
>  *  同步关系：比如说我跟小美去看电影，在此之前我们想买两瓶奶茶，因为我们前面有很多人买奶茶因此我们点完后我们就让店员做完后用vx给我们发个消息，我们先去买个爆米花上述我们能感受到生产和消费要有一定的顺序，消费完了再生产，生产完了再消费。所以生产者和消费者除了要保证临界资源的安全性外，还要保证消费过程中的合理性，所以消费者和供应商之间也应具备同步关系。

### 生产者消费者模型的特点： ###

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

**生产者消费者模型的特点如下：（321原则）**

*  三种关系：生产者和生产者（互斥），消费者和消费者（互斥），生产者和消费者（互斥、同步）
 *  两种角色：生产者和消费者（通常是由线程承担的）
 *  一个交易场所：通常是指内存中特定的一种内存结构（数据结构）

**生产者消费者模型的优点：**

1.  解耦
2.  支持并发，提高效率。例如生产者线程在缓冲区生产函数参数的时候，消费者线程也可以正常运行。这两个线程有原来的串行执行变为并发执行，提高效率
3.  支持忙闲不均

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

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

![image-20240101143744192][]

为了便于大家理解，我们以单生产者，单消费者，来进行讲解。

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

BlockQueue.hpp文件的代码逻辑如下：

如下我们把阻塞队列设计成BlockQueue模板类型，便于后续需要时的复用，该模板中的私有成员变量如下：

*  bq\_：用queue定义bq\_来承担阻塞队列的角色，
 *  cap\_：阻塞队列有容量大小，还需定义cap\_容量
 *  mutex\_：此bq\_阻塞队列将来是要被其它线程所访问的，这里的队列就充当一种需要被保护的全局变量，所以后续我们需要加锁，所以还需要在模板中定义一把锁mutex\_，来保护阻塞队列
 *  conCond\_（生产者条件变量）& proCond\_（消费者条件变量）：单纯的互斥锁会导致在生产过程中，生产者会出现条件不满足而导致的轮询检查频繁竞争锁从而导致另一方饥饿的问题，消费者也是如此。所以我们需要用到条件变量在双方条件不满足时让生产者进入不满足就休眠的状态，同样让消费者不满足条件也休眠的状态。等待条件满足了再唤醒对方。如上就是同步式的阻塞队列。

其内部公有成员函数如下：

*  BlockQueue构造函数：初始化cao\_容量为5（全局变量gDefaultCap），使用pthread\_mutex\_init动态初始化锁，使用pthread\_cond\_init动态初始化conCond\_（生产者条件变量）& proCond\_（消费者条件变量）
 *  ~BlockQueue析构函数：释放锁和两个条件变量
 *  push函数：供生产者线程向阻塞队列放数据，在生产数据前需要进行加锁，其次进行判断是否适合生产，bq\_阻塞队列里得有空间，满了就不生产，此时要进入休眠，等待你有空间时再将你唤醒，从而避免后续频繁的加锁解锁问题，不满就继续生产，生产完后唤醒消费者，因为可能上一次队列里没有数据，导致消费者进入休眠状态，生产后唤醒消费者来消费
 *  pop函数：供消费者线程向阻塞队列拿数据，在消费前需要加锁，其次进行判断是否适合消费，当bq\_阻塞队列为空时，不消费，进入休眠状态；当bq\_有数据时，唤醒，然后消费，每消费一个，就意味着当前阻塞队列空出一个位置，所以唤醒生产者来生产数据

私有成员函数：（为了更好的体现封装，我们对加锁、解锁、判断函数、等待函数、唤醒函数都进行了封装）

*  lockQueue加锁：复用pthread\_mutex\_lock函数
 *  unlockQueue解锁：复用pthread\_mutex\_unlock函数
 *  isEmpty判空：return bq\_.empty()即可，返回值bool类型。注意在主逻辑进行判断的时候一定要while(isEmpty())循环判断，而不能用if，因为wait函数被唤醒不一定代表条件是满足的，即使这种概率很小。
 *  isFull判满：return bq\_.size() == cap\_即可，返回值bool类型。注意在主逻辑进行判断的时候一定要while(isFull())循环判断，而不能用if，因为wait函数被唤醒不一定代表条件是满足的，即使这种概率很小。
 *  proBlockWait生产者等待：复用pthread\_cond\_wait函数即可，注意此函数要传入mutex\_锁，为的就是在阻塞线程的时候，帮我们自动释放mutex\_锁，防止占着锁不放，因为要维持生产者和消费者的互斥关系，传入的条件变量就是维护同步关系。注意：当我醒来的时候会重新获得mutex\_锁，然后才返回。从而避免了未持有锁而访问后续临界资源造成的安全问题。
 *  conBlockWait消费者等待：复用pthread\_cond\_wait函数即可，和上面一样，此函数的第一个参数条件变量维持的是生产者和消费者的同步关系，第二个参数锁维持的就是生产者和消费者的互斥关系。注意：当我醒来的时候会重新获得mutex\_锁，然后才返回。从而避免了未持有锁而访问后续临界资源造成的安全问题。
 *  wakeupPro唤醒生产者：复用pthread\_cond\_signal函数即可，此函数的目的是为了防止生产者因阻塞队列满了而即使后续有空位还依然处于等待状态的情况。要唤醒生产者生产数据供消费者消费
 *  wakeupCon唤醒消费者：复用pthread\_cond\_signal函数即可，此函数的目的是为了防止消费者因阻塞队列为空而即使后续有数据还依然处于等待状态的情况。要唤醒消费者消费，从而让生产者继续生产数据
 *  pushCore生产数据：复用push函数即可
 *  popCore消费数据：定义tmp临时变量保存队头数据，随后bq\_.pop()消费数据，返回此临时变量

**总代码如下：（BlockQueue.hpp文件）**

#pragma once
    #include <iostream>
    #include <queue>
    #include <cstdlib>
    #include <unistd.h>
    #include <pthread.h>
    using namespace std;
     
    // 定义默认容量大小为5
    const uint32_t gDefaultCap = 5;
     
    template <class T>
    class BlockQueue
    {
    public:
        // 构造函数
        BlockQueue(uint32_t cap = gDefaultCap)
            : cap_(cap)
        {
            pthread_mutex_init(&mutex_, nullptr);
            pthread_cond_init(&conCond_, nullptr);
            pthread_cond_init(&proCond_, nullptr);
        }
        // 析构函数
        ~BlockQueue()
        {
            pthread_mutex_destroy(&mutex_);
            pthread_cond_destroy(&conCond_);
            pthread_cond_destroy(&proCond_);
        }
     
    public:
        // 生产函数
        void push(const T &in) // const &: 纯输入
        {
            // 加锁
            // 判断 -> 是否适合生产 -> bq_是否满 -> 程序员视角的条件 -> 1、满(不生产) 2、不满(生产)
            // if(满) 不生产，休眠
            // else if(不满) 生产
            // 解锁
            lockQueue();
            while (isFull())
            {
                //before: 当我等待的时候，会自动释放mutex_
                proBlockWait(); // 阻塞等待，等待被唤醒
                //after: 当我醒来的时候，我是在临界区里醒来的
            }
            // 条件满足，可以生产
            pushCore(in); // 生产完成
            // 解锁
            unlockQueue();
            // 生产数据后唤醒消费者消费数据
            wakeupCon();
        }
        // 消费接口
        T pop()
        {
            // 加锁
            // 判断 -> 是否适合消费 -> bq_是否为空 -> 程序员视角的条件 -> 1、空(不消费) 2、有(消费)
            // if(空) 不消费，休眠
            // else if(有) 消费
            // 解锁
            lockQueue();
            if (isEmpty())
            {
                conBlockWait(); // 阻塞等待，等待被唤醒
            }
            // 条件满足，可以消费
            T tmp = popCore();
            // 解锁
            unlockQueue();
            // 消费数据后唤醒生产者生产数据
            wakeupPro();
            return tmp;
        }
     
    private:
        // 加锁
        void lockQueue()
        {
            pthread_mutex_lock(&mutex_);
        }
        // 解锁
        void unlockQueue()
        {
            pthread_mutex_unlock(&mutex_);
        }
        // 判空
        bool isEmpty()
        {
            return bq_.empty();
        }
        // 判满
        bool isFull()
        {
            return bq_.size() == cap_;
        }
        // 生产者等待
        void proBlockWait() // 生产者一定是在临界区的
        {
            // 1、在阻塞线程的时候，会自动释放mutex_锁，维持生产者和消费者的互斥关系
            pthread_cond_wait(&proCond_, &mutex_);
            // 2、当阻塞结束，返回的时候，pthread_cond_wait，会自动帮我们重新获得mutex_锁，然后才返回
        }
        // 消费者等待
        void conBlockWait() // 阻塞等待，等待被唤醒
        {
            // 1、在阻塞线程的时候，会自动释放mutex_锁，维持生产者和消费者的互斥关系
            pthread_cond_wait(&conCond_, &mutex_);
            // 2、当阻塞结束，返回的时候，pthread_cond_wait，会自动帮我们重新获得mutex_锁，然后才返回
        }
        // 唤醒生产者
        void wakeupPro()
        {
            pthread_cond_signal(&proCond_);
        }
        // 唤醒消费者
        void wakeupCon()
        {
            pthread_cond_signal(&conCond_);
        }
        // 生产数据
        void pushCore(const T &in)
        {
            bq_.push(in);
        }
        // 消费数据
        T popCore()
        {
            T tmp = bq_.front();
            bq_.pop();
            return tmp;
        }
     
    private:
        uint32_t cap_;           // 容量
        queue<T> bq_;            // blockqueue阻塞队列
        pthread_mutex_t mutex_;  // 保护阻塞队列的互斥锁
        pthread_cond_t conCond_; // 让消费者等待的条件变量
        pthread_cond_t proCond_; // 让生产者等待的条件变量
    };

**BlockQueueTest.cc文件的逻辑如下：**

*  在主函数中我们创建一个生产者线程和一个消费者线程，让生产者不断生产数据，让消费者不断消费数据，总代码如下：

#include "BlockQueue.hpp"
    #include <ctime>
    void *consumer(void *args)
    {
        
        //创建阻塞队列
        BlockQueue<int> *bqp = static_cast<BlockQueue<int>* >(args);
        while (true)
        {
        
            sleep(2);
            int data = bqp->pop();
            cout << "consumer 消费数据完成: " << data << endl;
        }
    }
    void *productor(void *args)
    {
        
        //创建阻塞队列
        BlockQueue<int> *bqp = static_cast<BlockQueue<int>* >(args);
        while (true)
        {
        
            //1、制作数据
            int data = rand() % 10;
            //2、生产数据
            bqp->push(data);
            cout << "productor 生产数据完成: " << data << endl;
            //生产慢一些
            // sleep(2);
        }
    }
     
    int main()
    {
        
        // 定义一个阻塞队列
        // 创建两个线程, productor, consumer
        // 建立联系 productor ———— consumer
        srand((unsigned long)time(nullptr) ^ getpid());
        BlockQueue<int> bq;
        pthread_t c, p;
        pthread_create(&c, nullptr, consumer, &bq);
        pthread_create(&p, nullptr, productor, &bq);
     
        pthread_join(c, nullptr);
        pthread_join(p, nullptr);
        return 0;
    }

**Makefile文件代码如下：**

blockQueue:BlockQueueTest.cc
    	g++ -o $@ $^ -lpthread -std=c++11
    .PHONY:clean
    clean:
    	rm -f blockQueue

生产者消费者步调一致：

![img][]

生产者生产的快，消费者消费的慢：![img][img 1]

#### 基于计算任务的生产者消费者模型（并发） ####

前面说到生产者消费者模型的其中一个优点是支持并发，可是我们上述的代码逻辑并没有体现到，我们上述代码的交易场所就是此queue队列，生产者生产数据，消费者消费数据都是在此queue队列里头依旧是互斥的啊，即使有同步，这里并不能深刻的感受到支持并发的特性，现在我们来更改上述代码，使其具有并发的特性。

上述BlockQueue模板中，我们使用的是int类型的数据，那么这里我们也可以用自己封装的类型，包括任务。假设这里的生产消费模型要能够支持完成计算的任务（生产是生产计算任务，消费者计算任务）。所以这里我们只需要定义一个Task类，内部需要包含一个run成员函数，并把此成员函数设计成仿函数。该函数代表着我们想让消费者如何处理拿到的数据。内部还需要提供一个get函数，从而辅助我们后续需要获得三个参与计算的操作数，这里巧用c++的引用实现

***\*Task.hpp文件的代码如下：\****

#pragma once
    #include <iostream>
    #include <string>
     
    class Task
    {
        
    public:
        Task()
            : elemOne_(0), elemTwo_(0), operator_('0')
        {
        }
        Task(int one, int two, char op)
            : elemOne_(one), elemTwo_(two), operator_(op)
        {
        }
     
        // 仿函数
        int operator()()
        {
        
            return run();
        }
     
        // 执行任务
        int run()
        {
        
            int result = 0;
            switch (operator_)
            {
        
            case '+':
                result = elemOne_ + elemTwo_;
                break;
            case '-':
                result = elemOne_ - elemTwo_;
                break;
            case '*':
                result = elemOne_ * elemTwo_;
                break;
            case '/':
            {
        
                if (elemTwo_ == 0)
                {
        
                    std::cout << "div zero, abort" << std::endl;
                    result = -1;
                }
                else
                {
        
                    result = elemOne_ / elemTwo_;
                }
            }
            break;
            case '%':
            {
        
                if (elemTwo_ == 0)
                {
        
                    std::cout << "mod zero, abort" << std::endl;
                    result = -1;
                }
                else
                {
        
                    result = elemOne_ % elemTwo_;
                }
            }
            break;
            default:
                std::cout << "非法操作: " << operator_ << std::endl;
                break;
            }
            return result;
        }
        // 获取参与计算的三个操作数
        int get(int &e1, int &e2, char &op)
        {
        
            e1 = elemOne_;
            e2 = elemTwo_;
            op = operator_;
        }
     
    private:
        int elemOne_;
        int elemTwo_;
        char operator_; // 具体的运算符号
    };

BlockQueueTest.cc文件的代码逻辑：

此时生产者放入阻塞队列的数据就是一个Task对象，我们利用rand函数随机生成两个运算变量one、two，再利用rand函数随机生成一个运算符号op。随后将这三者传入Task 的任务对象里。随后在内部打印一些提示信息。  
此时消费者从阻塞队列里头拿任务，拿到任务后直接利用先前的仿函数去执行任务，并定义result变量获得执行完任务后的结果，随后调用get函数获得参与计算的三个操作数。最后在内部打印一些提示信息。  
BlockQueueTest.cc文件的代码：

#include "Task.hpp"
    #include "BlockQueue.hpp"
    #include <ctime>
     
    const std::string ops = "+-*/%"; // 定义符号变量
     
    void *consumer(void *args)
    {
        
        // sleep(2);
        // 创建阻塞队列
        BlockQueue<Task> *bqp = static_cast<BlockQueue<Task> *>(args);
        while (true)
        {
        
            sleep(1);
            //1、消费任务
            Task t = bqp->pop();
            //2、处理任务
            int result = t();
            //获取参与计算的三个操作数
            int one, two;
            char op;
            t.get(one, two, op);
            cout << "consumer[" << pthread_self() << "]" << (unsigned long)time(nullptr) << 
            " 消费了一个任务: " << one << op << two << "=" << result << endl;
        }
    }
    void *productor(void *args)
    {
        
        // 创建阻塞队列
        BlockQueue<Task> *bqp = static_cast<BlockQueue<Task> *>(args);
        while (true)
        {
        
            sleep(2);
            // 1、制作任务
            int one = rand() % 50;
            int two = rand() % 20;
            // 利用rand函数让op计算符号随机设定
            char op = ops[rand() % ops.size()];
            Task t(one, two, op);
            // 2、生产任务
            bqp->push(t);
            cout << "productor[" << pthread_self() << "]" << (unsigned long)time(nullptr) << 
            " 生产了一个任务: " << one << op << two << "=?" << endl;
        }
    }
     
    int main()
    {
        
        srand((unsigned long)time(nullptr) ^ getpid());
        BlockQueue<Task> bq;
        pthread_t c, p;
        pthread_create(&c, nullptr, consumer, &bq);
        pthread_create(&p, nullptr, productor, &bq);
     
        pthread_join(c, nullptr);
        pthread_join(p, nullptr);
        return 0;
    }

![img][img 2]

*  我们不能简单的把生产者消费者模型理解为把数据或任务放在队列里，然后你来拿，制作任务和处理任务都要花费时间。当你放任务的时候，消费者可能正在处理任务，生产者生产任务的时候和消费者消费任务的是并发执行的。
 *  并发并不是在临界区中并发（一般而言），而是生产前（before blockqueue）和消费后（after blockqueue）对应的并发。我在处理任务时生产者可以不断的生产任务。这才是并发的。
 *  生产者消费者模型的解耦就体现在生产者和消费者之间不直接交互，而是通过一个中介（对应上述的阻塞队列）来帮忙交互。
 *  生产者消费者模型支持的忙闲不均指的是制作任务和处理任务，当制作任务要花1s，消费任务要话10s，那么生产者就可以多搞几个一同制作任务，然后放到仓库供消费者消费

ad\_join(c, nullptr);  
pthread\_join(p, nullptr);  
return 0;  
\}

[外链图片转存中...(img-dAaBjuqW-1707458273411)]
    
    - 我们不能简单的把生产者消费者模型理解为把数据或任务放在队列里，然后你来拿，制作任务和处理任务都要花费时间。当你放任务的时候，消费者可能正在处理任务，生产者生产任务的时候和消费者消费任务的是并发执行的。
    - 并发并不是在临界区中并发（一般而言），而是生产前（before blockqueue）和消费后（after blockqueue）对应的并发。我在处理任务时生产者可以不断的生产任务。这才是并发的。
    - 生产者消费者模型的解耦就体现在生产者和消费者之间不直接交互，而是通过一个中介（对应上述的阻塞队列）来帮忙交互。
    - 生产者消费者模型支持的忙闲不均指的是制作任务和处理任务，当制作任务要花1s，消费任务要话10s，那么生产者就可以多搞几个一同制作任务，然后放到仓库供消费者消费

[c42987e53ca7b5316882a596f5b6684e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/25/d08a6e52274540cd9b1420d8df45e1c1.png
[image-20240101143744192]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/25/230f8c4be9bf489789e74ce58321d815.png
[img]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/25/82f60c7ed59b4aaaa7926f761a7690f5.gif
[img 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/25/4a848faa78ca465b97eb2d051ff92264.gif
[img 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/25/3e8ae1c055024838966839e5434002ca.png