【Linux系统】结合有趣的小故事让你学懂生产者消费者模型

由故事引入模型

故事背景

有一个小朋友叫小C，他住的地方没有超市，只有几家供货商，因为每家供货商类型单一，买东西还要跑来跑去的，而且供货商晚上还不开门，买东西特别不方便，不仅小C觉得麻烦，其他人也觉得麻烦。小C想：为什么不能把这几家供货商的东西先放在一个地方呢，再由几个人专门卖，需要什么就直接挑选就好了，不用跑来跑去的，营业时间甚至可以全天。于是乎，小C就打电话给了市长，提了这个建议。市长知道了这个地方的市民买东西特别不方便，就接受了这个建议，于是就在这个地方建了个超市。
从此以后，小C和市民们买东西变得方便了，几家供货商把各种类型的商品送进超市，市民们只需要在超市进行挑选就可以了。大大节省了市民的时间，供应商也提高了工作效率，一次生产大批量的货物送进超市就好了，在货物充足时，供应商也能得到很好的休息，等货物缺乏再送过去。

由故事抽象出来的模型：
小C和市民们都是消费者，而供应商是生产者，超市是一种交易场所，为第三方
这就是生产者消费者模型，计算机中，生产者和消费者都是线程，第三方是一种特定数据结构的缓冲区。

线程之间想要通信，缓冲区一定要被所有线程看到， 也就是说缓冲区一定会被多线程并发访问， 那么缓冲区就要保护共享资源的安全，维护线程互斥与同步的关系。

供货商们的矛盾

由于超市的空间是有限的，这让供货商之间开始慢慢较量了，谁都想让自己的货物在超市多放一点。一天，供货商小S和供货商小D同时来超市放置自己的货物了。刚好超市这天只能放一家供货商的货物了，于是小S和小D就吵起来了。小S：“这块地方只有我能放货物，你不能放”。小D不服了：“凭什么只有你能放，我不能放？”于是两家供货商就大吵大闹，闹得沸沸扬扬的，不过这也不是一天两天的事了。超市知道了这件事后，就制定了一个叫做“锁”的规则：我这里有一把象征性的锁和钥匙，每天，谁能先拿到锁，谁就先放货物，放完后就解锁，下一次你们再继续竞争这把锁。

供货商是生产者，那么生产者和生产者之间的关系是竞争的关系。再极端一点，在线程中，我们叫互斥关系，同一时间一次只能执行一个线程。

市民们和供货商之间的矛盾一

小C早上想去超市买几箱可乐，很不巧超市没可乐了。于是小C过了一两小时又去超市问有可乐了吗，超市说没有。再过几个小时，小C再去，还是没可乐，过一会又去，还是没有。超市见小C频繁地来也不是个办法，就想了一个办法：你不要频繁的来了，你给我你的联系方式，等供货商送货来了，我再打电话给你。小C答应了这种请求。
超市想起前几个星期，超市货满放不下货物的时候，供货商也频繁地送货物来，每次都灰溜溜地回去了。于是超市也打电话对供货商说：你不要频繁地来了，你给我你的联系方式，等货物缺了，我再打电话给你，你再来。

小C想买可乐，但是超市没货，却隔一会就来问超市有货物吗。
供货商想送货进超市，但是超市货满了，却隔一会就问超市能进货了吗
这种可以抽象成线程的的频繁检测。
超市想出来的方案：等有货了再联系小C，等没货了再联系供应商。
可以抽象成缓冲区维护了生产者和消费者的同步关系，维护了线程之间的同步关系，让线程之间对第三方不再频繁的检测。

市民们和供货商之间的矛盾二

小C终于能去超市买可乐了，此时供货商小S想在这个地方放货物。由于超市空间限制，只能一个人在这里。小C："让我先买东西，你再放。"供货商小S又不服气了：“上次是我的同行和我抢，这次怎么到你了？，让我先放”。两个人谁也不服谁。于是，“锁”规则又可以用起来了。

小C是消费者，供货商是生产者，生产者和消费者之间也有互斥关系。

市民们的矛盾

小C好不容易能买可乐了，可是小N来了，他也想买这几箱可乐。
于是小C又和小N吵起来了，之前制定的“锁”规则又起效果了。

小C和小N都是消费者，消费者和消费者之间是“互斥关系”

模型总结

• 三种关系：生产者和生产者之间的关系（互斥），生产者和消费者之间的关系（互斥与同步），消费者和消费者之间关系（互斥）
• 两种角色：生产者和消费者
• 一个交易场所：通常是缓冲区

生产者消费者模型

为什么要使用生产者消费者模型？

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

生产者消费者模型的特点

由上面的故事已经进行总结。
生产者消费者模型是多线程同步与互斥的一个经典场景，其特点如下：

• 三种关系： 生产者和生产者（互斥关系）、消费者和消费者（互斥关系）、生产者和消费者（互斥关系、同步关系）。
• 两种角色： 生产者和消费者。（通常由进程或线程承担）
• 一个交易场所： 通常指的是内存中的一段缓冲区。（可以自己通过某种方式组织起来）

在编写生产者消费者代码的时候，本质就上就是对三种特点进行维护。

生产者和生产者、消费者和消费者、生产者和消费者，它们之间为什么会存在互斥关系？

介于生产者和消费者之间的容器可能会被多个执行流同时访问，因此我们需要将该临界资源用互斥锁保护起来。

其中，所有的生产者和消费者都会竞争式的申请锁，因此生产者和生产者、消费者和消费者、生产者和消费者之间都存在互斥关系。

生产者和消费者之间为什么会存在同步关系？

• 如果让生产者一直生产，那么当生产者生产的数据将容器塞满后，生产者再生产数据就会生产失败。
• 反之，让消费者一直消费，那么当容器当中的数据被消费完后，消费者再进行消费就会消费失败。

虽然这样不会造成任何数据不一致的问题，但是这样会引起另一方的饥饿问题，是非常低效的。我们应该让生产者和消费者访问该容器时具有一定的顺序性，比如让生产者先生产，然后再让消费者进行消费。

注意： 互斥关系保证的是数据的正确性，而同步关系是为了让多线程之间协同起来。

生产者消费者模型优点

• 解耦（生产者只负责生产，消费者只负责消费者）
• 支持并发
• 支持忙闲不均。· 假设没有缓冲区，且消费者和生产者的速度不匹配，则会造成CPU的浪费。生产者/消费者模型使得生产者/消费者的处理能力达到一个动态的平衡。

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

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

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

为了便于理解，这里以单生产者，单消费者为例。

先创建一个BlockingQueue类来充当我们的缓冲区。

#pragma once
#include <iostream>
#include <queue>

const int gcap = 5;//定义为 5方便后面进行测试
template <class T>
class BlockingQueue
{
public:
    BlockingQueue(const int cap = gcap) : _capapacity = cap
    {
    }
    !BlockingQueue()
    {
    }

private:
    std::queue<T> _q;//队列
    int _capacity;//队列的容量上限
};

生产者消费者模型是用在多线程场景下的，所以要我们要保证它是线程安全的，要保证线程互斥和线程同步。所以要加上锁和条件变量

• 在这个模型中，由于我们要避免生产者和消费者同时访问一份资源，只需要一把锁就够了。
• 但是条件变量需要两个。我们的要求是：当队列为空时，从队列中获取元素会被阻塞，直到队列中放入了元素；当队列为满时，往队列里存放元素也会被阻塞，直到队列里有元素被取出。所以一个条件变量是不够的，需要两个条件变量，分别表示满和空。

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

const int gcap = 5;//定义为5方便后面进行测试
template <class T>
class BlockingQueue
{
public:
    BlockingQueue(const int cap = gcap) : _capacity(cap)
    {
        pthread_mutex_init(&_mutex,nullptr);//初始化锁
        pthread_code_init(&_full,nullptr);//初始化条件变量
        pthread_code_init(&_empty,nullptr);//初始化条件变量
    }
    ~BlockingQueue()
    {
        pthread_mutex_destroy(&mutex);
        pthread_cond_destroy(&_full);
        pthread_cond_destroy(&_empty);
    }

private:
    std::queue<T> _q;//队列
    int _capacity;//队列的容量上限
    pthread_mutex_t _mutex;//定义锁
    pthread_cond_t _full;//条件变量,满时生产者阻塞
    pthread_cond_t _empty;//空时消费者阻塞
};

判空和判满函数

    bool isFull(){return _q.size() == _capacity;}
    bool isEmpty(){return _q.empty();}

首先先粗略写一下生产者要完成的任务：往容器里面放元素。这个时候需要判断容器是否是满的。

    void push(const T& in)//生产者把元素放进容器
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        if(isFull())//判断容器是否为满
        {
            //如果满了就进行等待
            pthread_cond_wait(&_full,&_mutex);
        }
        _q.push(in);//未满，就生产，放进容器
        pthread_mutex_unlock();
    }

这里简单谈一下pthread_cond_wait这个函数

• 我们只能在临界区内部，判断临界资源是否就绪，这就注定了在我们在当前一定是持有锁的。
• 要让线程进行休眠等待，就不能持有锁等待。
• 这就说明，pthread_cond_wait 要有锁的释放能力。
• 当线程醒来的时候，会继续从临界区内部继续运行，因为是在临界区被切走的。
• 注定了当线程被唤醒的时候，继续在pthread_cond_wait 函数向后运行，又要重新申请锁，申请成功才会返回

接下来再粗略写一下消费者要做的事情：从容器取元素。如果容器为空就等待。

    void pop(T* out)
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        if(isEmpty())//判断容器是否为空
        {
            pthread_cond_wait(&_empty,&_mutex);//如果为空，消费者就进行等待
        }
        *out = _q.front();//不为空，就取队列头部元素
        _q.pop();//取出以后，队列弹出该元素
        pthread_mutex_unlock(&_mutex);
    }

这两段生产者和消费者各自执行各自任务的代码是有问题的。

• 假如生产者要往容器存数据的时候，判断容器是满的，那么就去等待了。
• 此时消费者继续消费，当消费到容器为空时，消费者又去等待了。
  此时问题就是，没人能唤醒生产者和消费者。
  解决方法如下：

互相唤醒对方

• 当生产者能生产时，每次都使用函数唤醒消费者。
• 当消费者能消费时，每次都使用函数唤醒生产者。

代码如下：

    void push(const T& in)//生产者把元素放进容器
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        if(isFull())//判断容器是否为满
        {
            //如果满了就进行等待
            pthread_cond_wait(&_full,&_mutex);
        }
        _q.push(in);//未满，就生产，放进容器
        //此时可以加一些策略，比如容量为多少时就唤醒，我们这里就不加了。
        pthread_cond_signal(&_empty);
        pthread_mutex_unlock(&_mutex);
    }
    void pop(const T* out)
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        if(isEmpty())//判断容器是否为空
        {
            pthread_cond_wait(&_empty,&_mutex);//如果为空，消费者就进行等待
        }
        *out = _q.front();//不为空，就取队列头部元素
        _q.pop();//取出以后，队列弹出该元素
        pthread_cond_signal(&_full);
        pthread_mutex_unlock(&_mutex);
    }

目前代码如下：

#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
using namespace std;
const int gcap = 5;//定义为5方便后面进行测试
template <class T>
class BlockingQueue
{
public:
    BlockingQueue(const int cap = gcap) : _capacity(cap)
    {
        pthread_mutex_init(&_mutex,nullptr);//初始化锁
        pthread_cond_init(&_full,nullptr);//初始化条件变量
        pthread_cond_init(&_empty,nullptr);//初始化条件变量
    }
    bool isFull(){return _q.size() == _capacity;}
    bool isEmpty(){return _q.empty();}
    void push(const T& in)//生产者把元素放进容器
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        if(isFull())//判断容器是否为满
        {
            //如果满了就进行等待
            pthread_cond_wait(&_full,&_mutex);
        }
        _q.push(in);//未满，就生产，放进容器
        //此时可以加一些策略，比如容量为多少时就唤醒，我们这里就不加了。
        pthread_cond_signal(&_empty);
        pthread_mutex_unlock(&_mutex);
    }
    void pop(T* out)
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        if(isEmpty())//判断容器是否为空
        {
            pthread_cond_wait(&_empty,&_mutex);//如果为空，消费者就进行等待
        }
        *out = _q.front();//不为空，就取队列头部元素
        _q.pop();//取出以后，队列弹出该元素
        pthread_cond_signal(&_full);
        pthread_mutex_unlock(&_mutex);
    }
    ~BlockingQueue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_full);
        pthread_cond_destroy(&_empty);
    }

private:
    std::queue<T> _q;//队列
    int _capacity;//队列的容量上限
    pthread_mutex_t _mutex;//定义锁
    pthread_cond_t _full;//条件变量,满时生产者阻塞
    pthread_cond_t _empty;//空时消费者阻塞
};

小测试

我们写个多线程代码测试一下

#include "block_queue.hpp"
#include <ctime>
#include <unistd.h>
using namespace std;

void* consumer(void* args)
{
    BlockingQueue<int>* bq = static_cast<BlockingQueue<int>*> (args);
    while(true)
    {
        //1.将数据从blockqueue中获取
        int data = 0;
        bq->pop(&data);
        //2.结合某种业务逻辑，处理数据
        //这里先打印一下
        cout << "consumer data : " << data << endl;
    }
}
void* producer(void* args)
{
     BlockingQueue<int>* bq = static_cast<BlockingQueue<int>*> (args);
     while(true)
     {
        //1.生产者要先通过某种渠道获得数据，可以让用户从标准输入输入，也可以从网络里读
        //这里我们简单处理一下，自己创建随机一些数据测试一下就行
        int data = rand() % 10 + 1;
        //2.将数据推送到blockqueue，完成生产过程
        bq->push(data);
        cout << "prodecer data : " << data << endl;//打印查看
     }
}
int main()
{
    srand((uint64_t)time(nullptr) % 100000);//测试要用的数据
    //这里是为了方便理解，先写成单生产单消费
    BlockingQueue<int>* bq = new BlockingQueue<int>();
    pthread_t c,p;//c是消费者线程，p是生产者线程
    pthread_create(&c,nullptr,consumer,bq);//让消费者和生产者看到同一份队列
    pthread_create(&p,nullptr,producer,bq);

    pthread_join(c,nullptr);
    pthread_join(p,nullptr);
    return 0;
}

我们先让消费者的线程sleep（1），让它消费慢一点。然后生产者正常生产。

我们会发现，因为我们最开始容量最大为5，所有生产者很容易就把容器塞满了。
塞满以后就阻塞了，轮到消费者消费一个，根据我们代码所写，每次消费后就去唤醒生产者。生产者生产了，又满了。又轮到消费者消费一个，消费者又唤醒生产者。所以会出现消费一个，生产一个的情况。
很容易观察到，消费者消费的时候每次都是从队列的头获得数据的。

接下来，我们让消费者正常消费，生产者线程sleep(1)，生产慢一点

void* producer(void* args)
{
     BlockingQueue<int>* bq = static_cast<BlockingQueue<int>*> (args);
     while(true)
     {
        sleep(1);
        //1.生产者要先通过某种渠道获得数据，可以让用户从标准输入输入，也可以从网络里读
        //这里我们简单处理一下，自己创建随机一些数据测试一下就行
        int data = rand() % 10 + 1;
        //2.将数据推送到blockqueue，完成生产过程
        bq->push(data);
        cout << "prodecer data : " << data << endl;//打印查看
     }
}

还是出现了生产一个消费一个的情况。
原因是最开始队列是空的，生产者生产慢了，消费者只能等待。等到生产者生产了一个以后，我们没加任何策略，只要生产了就唤醒消费者线程。然后消费者消费了。队列又空了，消费者又要等待生产者生产。

由这个小测试我们可以看到，我们成功地让多线程协同起来了。

细节1 线程被误唤醒的情况

现在是单生产者单消费者的情况。如果改成只有一个消费者，五个生产者，有没有可能出现生产者被误唤醒的情况？
答案是可能的。假设现在队列里的数据满了，而消费者唤醒生产者的线程不是pthread_cond_signal(),而是pthread_cond_broadcast()，一下子唤醒五个生产者。

这时候问题就来了，如果消费者只消费了一个数据就全部唤醒了五个生产者，这五个生产者之前都通过if语句判断通过在进行等待，唤醒时都会从箭头所指处继续执行代码。都会执行push语句，就可能超过队列的容量上限。
这只是被误唤醒的一个例子，实际中可能还要很多情况被误唤醒。所以我们就要避免这种情况。

解决方法： if语句改成while即可
被唤醒的时候再判断一下是否是满了，满了继续等待，这样就不怕被误唤醒导致继续执行下面的代码了。

    void push(const T& in)//生产者把元素放进容器
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        while(isFull())//判断容器是否为满
        {
            //如果满了就进行等待
            pthread_cond_wait(&_full,&_mutex);
        }
        _q.push(in);//未满，就生产，放进容器
        //此时可以加一些策略，比如容量为多少时就唤醒，我们这里就不加了。
        pthread_cond_signal(&_empty);
        pthread_mutex_unlock(&_mutex);
    }

同理，消费者也必须改成while

    void pop(T* out)
    {
        pthread_mutex_lock(&_mutex);//加锁保证线程安全
        while(isEmpty())//判断容器是否为空
        {
            pthread_cond_wait(&_empty,&_mutex);//如果为空，消费者就进行等待
        }
        *out = _q.front();//不为空，就取队列头部元素
        _q.pop();//取出以后，队列弹出该元素
        pthread_cond_signal(&_full);
        pthread_mutex_unlock(&_mutex);
    }

细节2 生产者消费者模型高效在哪里？

生产者消费者模型就是生产者往容器里放元素，消费者再从容器里取元素。同时为了保证线程安全，我们还给它加锁了，所以是串行执行的，那么它高效在哪呢？

思考这几个问题：

• 生产者是不是也需要从外部获取数据才能送到容器？
• 消费者的数据是不是也要经过业务处理后才能送出去？
• 生产者什么时候获取数据的时候能干嘛？
• 消费者送出数据的时候能干嘛？

首先生产者需要从外部获取数据才能送到容器，在获取数据的同时也能把以前的数据送到容器。消费者要把处理后的数据送出去，送出去的同时也能从容器拿到新的数据。这就是生产者消费者模型高效的表现。

多生产者多消费者

我们可以接下来测试多生产多消费者的情况了，由于线程间是串行执行的，所以代码肯定是能执行的。

#include "block_queue.hpp"
#include <ctime>
#include <unistd.h>
using namespace std;

void* consumer(void* args)
{
    BlockingQueue<int>* bq = static_cast<BlockingQueue<int>*> (args);
    while(true)
    {
        sleep(1);
        //1.将数据从blockqueue中获取
        int data = 0;
        bq->pop(&data);
        //2.结合某种业务逻辑，处理数据
        //这里先打印一下
        cout << pthread_self() << " | "<<"consumer data : " << data << endl;
    }
}
void* producer(void* args)
{
     BlockingQueue<int>* bq = static_cast<BlockingQueue<int>*> (args);
     while(true)
     {
        sleep(1);
        //1.生产者要先通过某种渠道获得数据，可以让用户从标准输入输入，也可以从网络里读
        //这里我们简单处理一下，自己创建随机一些数据测试一下就行
        int data = rand() % 10 + 1;
        //2.将数据推送到blockqueue，完成生产过程
        bq->push(data);
        cout << pthread_self() << " | " << "prodecer data : " << data << endl;//打印查看
     }
}
int main()
{
    srand((uint64_t)time(nullptr) % 100000);//测试要用的数据
    //这里是为了方便理解，先写成单生产单消费
    BlockingQueue<int>* bq = new BlockingQueue<int>();
    pthread_t c1,c2,p1,p2;//c是消费者线程，p是生产者线程
    pthread_create(&c1,nullptr,consumer,bq);//让消费者和生产者看到同一份队列
    pthread_create(&c2,nullptr,consumer,bq);//让消费者和生产者看到同一份队列
    pthread_create(&p1,nullptr,producer,bq);
    pthread_create(&p2,nullptr,producer,bq);


    pthread_join(c1,nullptr);
    pthread_join(c2,nullptr);
    pthread_join(p1,nullptr);
    pthread_join(p2,nullptr);
    return 0;
}

使用ps -aL查看，包括线程在内，确实有五个线程在执行。

题外话

我们在测试的时候，只测试了int数据类型的

   BlockingQueue<int>* bq = new BlockingQueue<int>();

实际上，我们用的是一个类模板，也就说不仅仅可以传简单的数据类型，进行简单的数据处理，还可以传相应的类，类里面写你要接收的数据和处理数据的方式，然后由生产者从外界接受数据，存到对象里面，再把这个对象传给容器，消费者再拿出这个对象，根据类里面的处理数据的方式进行处理，然后在发到外界。文章来源地址https://www.toymoban.com/news/detail-585218.html

到了这里，关于【Linux系统】结合有趣的小故事让你学懂生产者消费者模型的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！