Linux系统编程——生产者消费者模型

一，模型介绍

1.1 预备知识（超市买东西的例子）

1.2 模型介绍

1.3 CP模型特点

二，基于阻塞队列的CP模型

2.1 介绍

2.2 阻塞队列的实现

2.3 主函数实现

2.4 效果展示

三，POSIX信号量

3.1 信号量原理

3.2 信号量概念

3.3 信号量操作函数

四，基于环形队列的CP模型

3.1 介绍

3.2 生产者和消费者申请和释放资源

3.3 环形队列的实现

3.4 主函数的实现

3.5 效果演示

一，模型介绍

1.1 预备知识（超市买东西的例子）

我（消费者）去超市买东西，但是超市并不生成商品，商品由对应的一系列供货商生产（生产者），而基于消费者和生产者之间的超市，作为一种容器，接收生产者产出的商品，同时也将商品对消费者进行提供。如下图：
问题：我为什么不直接去找供货商要东西呢？为了提高生产效率
生产者就不关注具体的用户需求，只研究超市想要什么，这样，供货商只关心它的生产，超市只负责如何把商品派发给消费者，消费者同理；所以在逻辑上完成了一次解耦，通过不同的角色进行解耦提高效率
而这个超市，对应到程序中，就是一个大的缓冲区，但是光有超市还不够，还得有对应的管理机制，然后“超市 + 管理”组合在一起就是我们的“生产者消费者模型”

1.2 模型介绍

生产者消费者模型，简称CP模型（consumer，productor）是多线程同步与互斥的一个经典场景，其主要包含下面三种特点：

三种关系：生产者和生产者（互斥关系）、消费者和消费者（互斥关系）、生产者和消费者（互斥与同步关系）。
两个角色：生产者和消费者。
一个交易场所：通常指内存中的一段缓冲区。

问题：三种关系为什么存在互斥关系？

解答：因为一个超市不止一个供货商和一个顾客在访问，也就是介于生产者和消费者之间的容器可能会被多个执行流访问，因此我们需要将该临界资源保护起来，于是就有了互斥关系。

问题：生产者和消费者为什么会有同步关系？

解答：很简单，一个超市无法出售没有的物品，也不能继续进货已经满载的物品；所以当容器为满时，生产者停止生产，当容器为空时，消费者停止消费，两个角色的执行是有一定先后顺序的，所以需要有同步关系

使用工程师思维，重新理解生产者消费者：

生产者和消费者就是由我们的线程承担 --> 给线程角色化，对于交易场所，就是某种数据结构表示的缓冲区，而商品就对应我们的数据

超市里有没有新增商品，生产者最清楚；超市里还剩多少空间让供货商生产，消费者最清楚 --> 生产者生产完了就可以通知消费者来消费了（通知消费者数据可以被读写了），消费者把数据拿走（空间又有了，就可以通知生产者继续生产了） --> 让消费者线程和生产者线程同步了

1.3 CP模型特点

主要有以下三点

实现生产与消费的解耦
支持并发
支持忙闲不均

生产者只生产数据，消费者只消费数据，二者可以在一定基础上并发进行，好比你在超市买牛奶的时候，对应的厂家的工厂也可以在生产牛奶，这就是一种解耦

二，基于阻塞队列的CP模型

2.1 介绍

在多线程中，阻塞队列（Blocking Queue）是一种常用于实现生产者消费者模型的一种数据结构，如下图：

生产和消费的过程不仅仅是把生产者把数据放队列里，然后消费者来拿，有下面两个问题：

①生产者的数据哪来的呢？ ②消费者如何使用发送过来的数据呢？

我不知道从哪来，但是生产数据一定要花时间，而且消费者使用数据也一定要花时间，所以这个模型支持并发是支持生产者生成的过程和消费者消费的过程是“并发”的，一个访问临界区代码，一个访问非临界区代码，所以网上很多人说CP模型“高效”，所以“高效”不是指生产者和消费者两个互磕访问各自临界区，而是有一定概率“并发”执行，在这种情况下，CP模型才是高效的

2.2 阻塞队列的实现

我们先来看下阻塞队列的实现代码：

lockGuard.hpp头文件就是我们之前写的RAII的加锁方式：

lockGuard.hpp：

#pragma once#include <pthread.h>class Mutex
{
public:Mutex(pthread_mutex_t *lock): _lock(lock){}~Mutex(){}void Lock(){pthread_mutex_lock(_lock);}void UnLock(){pthread_mutex_unlock(_lock);}private:pthread_mutex_t *_lock;
};class LockGuard // RAII
{
public:LockGuard(pthread_mutex_t *lock): _mutex(lock){_mutex.Lock();}~LockGuard(){_mutex.UnLock();}private:Mutex _mutex;
};

然后是阻塞队列代码的实现：

// 阻塞队列，当消费者去读取数据时，如果队列为空，消费者就阻塞者；生产者同理，当队列满了，生产者也阻塞住，不生产任务
#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
#include <mutex>
#include "lockGuard.hpp"const int gDefaultCap = 5;template <class T>
class BlockQueue // 我们可以往阻塞队列里放数据让消费者去拿
{
public:BlockQueue(int capacity = gDefaultCap): _capacity(gDefaultCap){// 构造时初始化锁的条件变量pthread_mutex_init(&_mtx, nullptr);pthread_cond_init(&_Empty, nullptr);pthread_cond_init(&_Full, nullptr);// low_water = _capacity / 3;// high_water = (_capacity * 2) / 3;}void push(const T &in){LockGuard lockguard(&_mtx); // 自动调用构造，自动加锁，然后函数结束后自动析构// 1，先检测当前临界资源是否满足访问条件while (_bq.size() == _capacity) // 如果队列满了，就阻塞住，Full条件变量由pop控制，当消费者消费数据也就是pop时，生产者才开始生产pthread_cond_wait(&_Full, &_mtx);// 检测临界资源其实也是在访问临界资源，所以需要在临界区中，但是这时候我是持有锁的，这时候我等待了，锁没有释放，这时候消费者就无法拿到锁进行消费// pthread_cond_wait接口的第二个参数是一个锁，表示成功调用wait并阻塞之后，传入的锁会自动释放，当我被唤醒时，从被阻塞挂起的位置唤醒// 当我们被唤醒的时候，pthread_cond_wait会自动给线程获取锁，且这个过程是原子的// 但是wait也是一个函数，只要是函数调用，就有可能“失败”，失败后就没有被阻塞了，所以可能存在伪唤醒情况，就是唤醒条件没满足，线程就被唤醒了，所以我们不能用if判断，应该用while判断// 2，访问临界资源，100%确定资源是就绪的_bq.push(in); // 条件彻底满足时，再让生产者往队列放数据// if (_bq.size() < low_water) // 如果任务数小于指定值，通知生产者赶紧来生产pthread_cond_signal(&_Empty);}void pop(T *out){LockGuard lockguard(&_mtx);while (_bq.size() == 0) // 如果队列为空，消费者就阻塞住，Empty条件变量由生产者控制，当生产者生产数据也就是push时，通知消费者开始消费数据pthread_cond_wait(&_Empty, &_mtx);*out = _bq.front();_bq.pop();// if (_bq.size() > high_water) // 如果队列里任务数大于指定值了，通知消费者赶紧来消费pthread_cond_signal(&_Full);}~BlockQueue(){pthread_mutex_destroy(&_mtx);pthread_cond_destroy(&_Empty);pthread_cond_destroy(&_Full);}private:std::queue<T> _bq;     // 阻塞队列，里面放模板，是生产者和消费者的 “共享资源”int _capacity;         // 表示阻塞队列的容量上限，表示队列当中的极值，与queue.size()无关pthread_mutex_t _mtx;  // 通过互斥锁保证队列安全，因为STL容器本来不是线程安全的，需要我们自己保护pthread_cond_t _Empty; // 条件变量，表示阻塞队列是否为空pthread_cond_t _Full;  // 条件变量，表示阻塞队列是否为满// int low_water = 0;// int high_water = 0;
};

结合代码和注释，介绍下代码：

阻塞队列的原始队列数据结构我们直接STL库中的queue的即可，并且将BlockingQueue中的数据使用模板<T>代替，方便以后需要时进行复用
这里需要用到两个条件变量，Full和Empty，按最简单的来说，当队列为空时，Empty就不就绪了，当队列为满时，Full就不就绪了
上述代码将队列的默认大小设为5，表示队列里最多只能存5个<T>数据，当队列里满的时候，生产者就阻塞住，就是等待Full条件变量就绪，Full条件变量由pop控制，当消费者消费数据也就是pop时，生产者才开始生产；队列为空时，与上面同理，Pop时阻塞住
当生产者生产完一个数据后，意味着阻塞队列里至少有一个数据，根据条件变量的知识，生产者生产完数据后就要“通知”消费者来消费；消费者同理，消费了一个数据后，也要通知生产者来生产数据

可以看到上述代码中间有一大块灰色区域，也就是一大块注释，其实是在解释生产消费时，检测队列里的数据量是否符合各自要求时，为什么要用while循环判断：

2.3 主函数实现

前面说过，我们用模板<T>充当阻塞队列里的数据，这样我们就可以往里面填充各种数据，而大部分生产者消费者模型里面的“数据”，本质是“任务”，再实际一点，就是“函数”：

Task.hpp：

#pragma once#include <iostream>
#include <functional>
typedef std::function<int(int, int)> func_t; // 设置一个函数类型，返回值为int，两个参数都为intclass Task
{
public:Task() {}Task(int x, int y, func_t func): _x(x), _y(y), _func(func){}int operator()(){return _func(_x, _y);}public:int _x;int _y;func_t _func;
};

然后是主函数的实现代码：

#include "BlockQueue.hpp"
#include "Task.hpp"
#include <pthread.h>
#include <unistd.h>
#include <ctime>
int myAdd(int x, int y)
{return x + y;
}
// 生产者消费者模型提高效率，消费线程和生产线程处理数据都需要时间的，但是生产者可以一直往仓库里放数据，消费者可以一直从仓库里拿数据，两个线程就实现了一定的并发
// 所以生产者消费者模型提高效率，更多的是利用缓冲区，提高消费线程和生产线程的并发度，并发不是拿任务时并发，而是处理任务时并发void *consumer(void *args)
{BlockQueue<Task> *bqueue = (BlockQueue<Task> *)args;while (true){// 获取任务Task t;bqueue->pop(&t);// std::cout << "消费一个数据：" << a << std::endl;// 完成任务std::cout << pthread_self() << " consumer: " << t._x << " + " << t._y << " = " << t() << std::endl;sleep(1);}return nullptr;
}void *productor(void *args)
{BlockQueue<Task> *bqueue = (BlockQueue<Task> *)args;while (true){// 制作任务 --> 以后这个任务可能从各种途径来int x = rand() % 10 + 1;usleep(rand() % 1000);int y = rand() % 5 + 1;Task t(x, y, myAdd); // 制作一个简单的加法任务// 生产任务bqueue->push(t);// std::cout << "生产一个数据：" << a << std::endl;// a++;std::cout << pthread_self() << " productor: " << t._x << " + " << t._y << " = ?" << std::endl;sleep(1);}return nullptr;
}int main() // 负责生产消费
{srand((uint64_t)time(nullptr) ^ getpid() ^ 0x32457);BlockQueue<Task> *bqueue = new BlockQueue<Task>();pthread_t c[2], p[2];pthread_create(c, nullptr, consumer, bqueue); // 线程传参是可以传递对象的pthread_create(c + 1, nullptr, consumer, bqueue);pthread_create(p, nullptr, productor, bqueue);pthread_create(p + 1, nullptr, productor, bqueue);pthread_join(c[0], nullptr);pthread_join(c[1], nullptr);pthread_join(p[0], nullptr);pthread_join(p[1], nullptr);delete bqueue;return 0;
}

解释下代码：

创建4个线程，两个生产者两个消费者，上面的代码是生产与消费的时间一致，因为方便模拟现象。前面说过，我不知道生产者的数据从哪里来，但是一定要花时间，消费也是，所以可以改变上面代码的sleep休眠时间，来模拟生产消费时间不同的情况，这里就不测试了哈，小伙伴们可以自行复制测试哈~
关于Task类，使用了C++11的包装器，包装了一个两个参数为int，返回值为int的函数，结合类的构造与重载，最终组成上述代码的形式

2.4 效果展示

上面的是最基本的情况，就是生产者一生产一个任务，消费者立马就去读取并且执行了。

那么我们可以先让生产者一下子把任务生产完，也就是填满队列，这是生产者会阻塞住，只需要把主函数的productor里的sleep(1); 注释掉即可：

可以看到生产者一下子就生产了5个任务，后续就是消费者消费一个生产者就生产一个，消费者同理，把consumer的sleep(1); 注释掉就可以模拟消费者阻塞的场景了

三，POSIX信号量

关于信号量的基本在“进程间同学”博客最后已经讲过，如下链接：Linux系统编程——进程间通信（管道与共享内存）_共享内存管道-CSDN博客

3.1 信号量原理

我们讲可能被多个执行流同时访问的资源叫做“临界资源”，临界资源需要进行保护，不然会出现数据不一致的问题
当我们仅用一个互斥锁对临界资源进行保护时，相当于我们的操作单位是整个临界资源，效率会比较低下
但实际上我们可以将共享资源看做成多份，假设一个数组被分成三个部分，每个部分都允许一个线程来访问，这样就可以实现多线程同步访问，那么分成多少份决定了允许多少个线程进来，而完成这步操作的技术保障就是“信号量”

3.2 信号量概念

电影院例子，买票的本质：资源（座位）的预定机制。
信号量本质是一个“计数器”描述临界资源数目，我们访问临界资源的时候，必须先申请信号量资源（sem--，预定资源 -> P操作），使用完毕信号量资源之后要归还信号量资源（sem++，释放资源 -> V操作）
问题：这个“计数器”的本质是什么？临界资源的数量。所以我们在访问临界资源前要先申请信号量，而申请信号量之后就不用再判断临界资源是否就绪了，所以这个“计数器”的本质是“用来描述临界资源数目的，并且把资源是否就绪的判断放在了临界区之外”，所以申请信号量时，就已经间接地判断临界资源是否就绪了

共享资源 -> 任何一个时刻都只有一个执行流在进行访问 -> 临界资源，临界区
所以有了互斥，只能有一个去访问，潜台词就是共享资源是被当作整体使用的（相当于一个电影院厅只能让一个人去看电影），这样做是对的，但是不合理，效率太低，所以我们可以想办法让多个线程访问临界资源的不同位置（一个电影院厅有很多个座位，每个座位位置不一样） --> 让不同的执行流访问不同区域的话，就可以继续并发了，只有访问共享资源的同一个位置的时候，我们再进行同步和互斥
问题：①你咋知道一共有多少个资源，还剩多少个？ ②你咋保证这个资源就是给你用的？（程序员编码解决）我咋知道我一定可以具有一个共享资源的呢？（信号量）

总结：信号量是一个保证PV操作的原子性的一把计数器。

3.3 信号量操作函数

信号量操作有四个：初始化，P操作，V操作，销毁。

初始化信号量

第一个参数sem就是需要初始化的信号量，第二个参数为0时，表示该信号量线程间共享，非零表示进程间共享，第三个参数就是初始化信号量时的初始值（计数器的初始值）

P操作

P操作，使信号量减减

V操作

V操作，使信号量加加

销毁信号量

释放信号量空间

四，基于环形队列的CP模型

3.1 介绍

在阻塞队列中，进行资源操作是以整个队列为单位进行操作的，这样效率怎么说都是有点底下的

所以我们可以用另外一种对策，也就是通过编码，使每个线程去访问队列的一部分，于是诞生了环形对了的CP模型，它对数据的判断做了进一步的解耦：生产者关注空间资源，消费者关注数据资源，使双方访问线程的一部分，一定程度上能提高整体效率，至少加锁解锁的区域不再是整体了：

生产者关注的是环形队列中是否有空间（space），只要有空间生产者就可以进行生产
消费者关注的是环形队列中是否有数据（data），只要有数据就可以消费

如上图所示，我们用到的是一个首尾相连的一个数组，这个数据有两个指针，刚开始指向同一个位置，当生产者生产一个数据后，对一个位置进行标记，也就是填充了数据“data”，然后指针往后移，继续生产；消费者同理，消费了一个数据，也对该位置进行标记，将原来的“data”标记变为“space”，然后指针也往后移。

所以，环形队列的CP模型里的生产消费，本质上就是两个指针的“追逐游戏”。但是关于这个“游戏”，有两条规则绝对不能被打破或者触发：

规则一：生产者和消费者不能对统一位置进行访问：

规则二：双方指针不能相遇，或者都不应该将对方套一个圈以上

当消费数据效率比生产数据慢时，可能会出现上图左边的情况，如果这种情况下再次生产，会覆盖掉未被消费的数据，这是不被允许的，所以不能此时生产者不能再继续生产
右图同理，当消费指针超过生产指针时，会使数据“二次访问”，这也是绝对不允许的

其实上面说了大半天，在实际编写代码中不要考虑那么多，因为我们会用到“信号量”的相关内容，信号量的使用，能更方便我们编写代码

3.2 生产者和消费者申请和释放资源

生产者申请空间资源，释放数据资源

生产者每次生产数据前都要申请信号量space_sem:

如果space_sem不为0，则申请成功，生产者继续往下执行生产操作
如果space_sem为0，申请失败，生产者阻塞住，等待被唤醒
当生产者生产完数据后，需要对data_sem信号量进程V操作，data_sem++，表示队列里多了一个数据

消费者申请数据资源，释放空间资源

如果data_sem不为0，则申请成功，消费者继续往下执行消费操作
如果data_sem为0，申请失败，消费者阻塞住，等待被唤醒
当消费者消费完数据后，需要对space_sem信号量进程V操作，space_sem++，表示队列里多少了一个数据，多了一个空间

3.3 环形队列的实现

#ifndef _Ring_QUEUE_HPP_
#define _Ring_QUEUE_HPP_#include <iostream>
#include <vector>
#include <pthread.h>
#include <semaphore.h> //信号量头文件const int g_default_num = 5; // 设置环形队列大小template <class T>
class RingQueue
{
private:void P(sem_t &sem){sem_wait(&sem); //--，如果申请不成功，就挂起}void V(sem_t &sem){sem_post(&sem); //++}public:RingQueue(int default_num = g_default_num): ring_queue_(default_num), _num(default_num), c_step(0), p_step(0){// 初始化锁pthread_mutex_init(&clock, nullptr);pthread_mutex_init(&plock, nullptr);// 初始化信号量sem_init(&space_sem_, 0, default_num);sem_init(&data_sem_, 0, 0);}~RingQueue(){pthread_mutex_destroy(&clock);pthread_mutex_destroy(&plock);sem_destroy(&space_sem_);sem_destroy(&data_sem_);}// 为空或为满时，都只允许一个生产者或消费者来访问队列void push(const T &in) // 生产者：空间资源    问题：生产者们的临界资源是什么？ c_step下标，所以我们要加锁保护这个下标，消费者同理{// 问题：信号量是一种资源的预定机制，所以是先加锁还是先申请信号量？还是先申请信号量再加锁？// 1，技术角度：信号量资源不需要被保护，因为它本身就是原子的，不需要加锁；要尽量保证加锁解锁区域的代码小一些，能提高整体效率// 2，逻辑角度：先申请信号量再申请锁，可以在多线程申请时，使申请信号量和申请锁的时间变成并行的，能提高效率P(space_sem_); // 先申请空间资源，未申请成功就从这里挂起pthread_mutex_lock(&plock);// 走到这里一定是竞争成功的生产者 --> 就一个ring_queue_[p_step] = in;p_step++;p_step %= _num;pthread_mutex_unlock(&plock);V(data_sem_); // 生产了一个资源，就把资源信号量做V操作}void pop(T *out) // 消费者：数据资源{P(data_sem_); // 先申请数据资源pthread_mutex_lock(&clock);// 走到这里一定是竞争成功的消费者 --> 就一个*out = ring_queue_[c_step];c_step++;c_step %= _num;pthread_mutex_unlock(&clock);V(space_sem_); // 消费了一个数据，使空间++}private:std::vector<T> ring_queue_;int _num;int c_step; // 消费下标int p_step; // 生产下标sem_t space_sem_; // 空间信号量sem_t data_sem_;  // 数据信号量pthread_mutex_t clock; // 消费者的锁pthread_mutex_t plock; // 生产者的锁
};#endif

解释下代码：

环形队列通过vector实现，只要控制指针每次走到结尾时，通过%运算使指针回到开始即可完成闭环操作
生产者每次生产的数据会放到数组中p_step的位置，然后p_step++，消费者同理，消费完后c_pos++
即使信号量保证了绝大部分临界资源的安全，但是对于下标指针p_step和c_step，还是会有很多线程同时访问，比较生产者和消费都都可能是多个，所以要加锁保护。至于申请信号量和加锁的先后顺序，代码注释已经给出了解释

3.4 主函数的实现

include <cstdlib>
include <ctime>
include <sys/types.h>
include <unistd.h>
include <string>
include "ringQueue.hpp"
include "Task.hpp"
nt myAdd(int x, int y)
{return x + y;
}
truct ThreadData
{RingQueue<Task> *_rq;std::string threadname;
;
oid *consumer(void *args) // 消费
{sleep(1);ThreadData *td = static_cast<ThreadData *>(args);RingQueue<Task> *rq = td->_rq;std::string name = td->threadname;while (true){Task t;// 1，从环形队列中拿数据rq->pop(&t);// 2，进行一定的处理 --> 当然，也不要忽略它的时间消耗问题std::cout << name << ": 消费数据：" << t._x << " + " << t._y << " = " << t() << std::endl;sleep(1);}
}
oid *productor(void *args) // 生产
{ThreadData *td = static_cast<ThreadData *>(args);RingQueue<Task> *rq = td->_rq;std::string name = td->threadname;while (true){// 1，构建数据或者任务对象 --> 一般从外部来 --> 不要忽略构建任务或数据的时间问题int x = rand() % 10 + 1;usleep(rand() % 1000);int y = rand() % 5 + 1;Task t(x, y, myAdd); // 制作一个简单的加法任务std::cout << name << ": 生产数据：" << t._x << " + " << t._y << " = ?" << std::endl;// 2，推送到环形队列rq->push(t);sleep(1);}
}
nt main()
{srand((uint64_t)time(nullptr) ^ getpid());RingQueue<Task> *rq = new RingQueue<Task>(5); // 这个10可以自己修改// rq->debug();pthread_t c[3], p[2]; // 5个消费线程，3个生产线程for (int i = 0; i < 3; i++){ThreadData *td = new ThreadData;td->_rq = rq;td->threadname = "消费者-" + std::to_string(i);pthread_create(c + i, nullptr, consumer, (void *)td);}for (int i = 0; i < 2; i++){ThreadData *td = new ThreadData;td->_rq = rq;td->threadname = "生产者-" + std::to_string(i);pthread_create(p + i, nullptr, productor, (void *)td);}for (int i = 0; i < 3; i++)pthread_join(c[i], nullptr);for (int i = 0; i < 2; i++)pthread_join(p[i], nullptr);return 0;
}

解释下上面的代码：