【Linux探索学习】第三十二弹——生产消费模型：基于阻塞队列和基于环形队列的两种主要的实现方法

Linux学习笔记：

https://blog.csdn.net/2301_80220607/category_12805278.html?spm=1001.2014.3001.5482

前言：

在前面我们已经学习了关于线程的主要知识，包括线程的基础知识以及线程的同步与互斥等内容，今天我们来学几个线程知识的拓展内容和几个经典的应用场景，比如：生产消费模型（cp问题）、线程池、有关线程的单例模式等，下面我们就来看一下这几点内容（本篇为生产消费模型，线程池和单例模式在后面讲）

目录

1. 为什么要使用生产消费模型

2. 生产消费模型优点

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

3.1 简单实现

3.2 伪唤醒

3.3 拓展

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

4.1 环形队列

4.2 代码实现

5. 总结

1. 为什么要使用生产消费模型

我们来看这样一个例子

在一个超市购货和售货系统中，超市相当于临界于供货商和消费者之间的资源，一个超市不止有一个供货商，当然也不止有一个消费者，这些供货商和消费者之间正是因为超市的存在所以可以避免互相接触，供货商只需要将商品直接批发给超市就好了，消费者去超市货架上获取商品，这样就起到解耦的作用，从而提高了效率

生产消费模型正是如此，不同的是超市我们用一些特定结构的内存空间来表示，比如阻塞循环队列等，而供货商和生产者我们则是用线程来替代

既然我们说超市是用特定结构的内存空间来代替的，而且它还处于生产者和消费者之间，所以它本质上也是作为临界资源存在的，既然是临界资源就一定存在一定的同步与互斥问题，具体的关系如下图所示：

我们可以把这三种关系结合上面讲的生产消费模型的结构总结为“321原则”

2. 生产消费模型优点

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

3.1 简单实现

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

我们先来看一下我们基于阻塞队列建立的生产消费模型的基本格式：

#pragma once

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

template<class T>
class BlockQueue
{
static int defaultcap=1;
public:
BlockQueue(int maxcap=defaultcap):max_cap(maxcap)
{
pthread_mutex_init(&mutex_,nullptr);
pthread_cond_init(&p_cond_,nullptr);
pthread_cond_init(&c_cond_,nullptr);
}
void Push(const T& in)
{
//生产
}
T Pop()
{
//消费
}
~BlockQueue()
{
pthread_cond_destroy(&p_cond_);
pthread_cond_destroy(&c_cond_);
pthread_mutex_destroy(&mutex_);
}
private:
queue<T> q_; //共享资源
int max_cap; //最大容量
pthread_mutex_t mutex_; //只需要有一把锁就够了，在我们创建的这个简单的生产消费模型中不存在同时生产和消费的场景
pthread_cond_t p_cond_;
pthread_cond_t c_cond_; //创建了两个条件变量，因为生产者和消费者启动的条件不一样
};

在这里我们把这个阻塞队列作为共享资源，生产和消费都是在这个内存空间中对数据进行操作的，所以我们需要一把锁确保这个共享资源不被多次占用

下面我们看一下生产和消费的具体方法：

生产（Push）：

void Push(const T& in)
{
//生产者
pthread_mutex_lock(&mutex_);
if(q_.size()==max_cap)
{
pthread_cond_wait(&p_cond_,&mutex_); //当队列中没有数据时，激活条件变量同时把锁释放
}
q_.push(in);
//因为我们创建的是一个简单的生产消费模型，且最大容量为1，所以当我们生产一个数据后就达到了最大容量，需要激活消费者
pthread_cond_signal(&c_cond_);
pthread_mutex_unlock(&mutex_);
}

消费（Pop）：

T Pop()
{
//消费者
pthread_mutex_lock(&mutex_);
if(q_.size()==0)
{
pthread_cond_wait(&c_cond_,&mutex_);
}
T out=q_.front();
q_.pop();
pthread_cond_signal(&p_cond_);
pthread_mutex_unlock(&mutex_);
return out;
}

总体来说生产和消费的实现方法也是很简单的，抛开锁和信号量的使用来讲，就是很简单的往一个队列中插入和拿取数据，锁和信号量的使用方法在前面讲过，各位可以结合注释看一下如何用

以上就是我们基于阻塞队列的一个简单的生产消费模型，下面我们结合一段代码来使用一下我们的生产消费模型：

#include"BlockQueue.hpp"
#include<iostream>
#include<pthread.h>
#include<ctime>
#include<unistd.h>
using namespace std;

void *Productor(void *args)
{
//args是void*类型的，我们需要先转换
BlockQueue<int> *bq=static_cast<BlockQueue<int>*>(args);
//让我们的线程处于死循环的状态，即一直生产
while(true){
int in=rand()%10+1;

usleep(10); //模拟处理数据所需时间
bq->Push(in);
cout<<"Productor make a num: "<<in<<" threadid: "<<pthread_self()<<endl;
sleep(1); //让生产者在生产一个数据后睡眠一秒，便于观察打印结果
}
return nullptr;
}
void *Consumer(void *args)
{
BlockQueue<int> *bq=static_cast<BlockQueue<int>*>(args);
while(true){
int out=bq->Pop();
usleep(100); //模仿数据处理所需时间
cout<<"Consumer get a num: "<<out<<" threadid: "<<pthread_self()<<endl;
}
return nullptr;
}
int main()
{
srand(time(nullptr));
BlockQueue<int> *bq=new BlockQueue<int>();
pthread_t c,p;

//创建生产和消费两个线程，并且将bq作为参数传给两者
pthread_create(&p,nullptr,Productor,bq);
pthread_create(&c,nullptr,Consumer,bq);

//回收线程
pthread_join(c,nullptr);
pthread_join(p,nullptr);

delete bq;
return 0;
}

执行结果（截取部分）：

此时我们就可以看到我们的生产者创建一个数据，并将这个数据传给阻塞队列，也就是临界资源中，消费者从里面拿数据，同时也没有出现互斥等问题

3.2 伪唤醒

其实在生产者和消费者互相唤醒的过程中，我们有时候可能会出现这样的一种错误

比如如果我们此时有多个生产者，并且条件队列中有多个线程在等待，我们的消费者此时如果消费了一个数据，按理说就会唤醒等待队列最前面的线程获取锁资源并生产一个数据，但是如果我们在唤醒时采用的是全部唤醒的方式呢?那么此时等待队列中的线程都会被唤醒了，同时竞争锁，在第一个线程获得锁并将数据生产后，第二个线程仍会获得锁并生产数据，但是数据只消耗了一个，所以此时就会引发错误，我们把导致这种错误的原因叫做伪唤醒

那么如何解决伪唤醒的问题呢?

很简单，我们采用while循环的方式来进行判断，这样每一个线程在获取锁执行下一步时都要进行判断

3.3 拓展

上面我们讲的就是一个最简单的生产消费模型，但实际上生产消费模型可以在更多的场景下进行使用，比如在上面我们往循环队列中传的是整数，但是我们也可以创建任务传进去，这样消费者得到的就是各种任务

再比如我们上面创建的是单生产者和单消费者，但如果有多个执行者呢？我们该如何做呢？

很简单，我们直接创建就行，因为不管几个生产者和消费者，它们访问的都还是同样的资源，所以还是只需要一把锁就足够了，并不需要改动什么，最核心的原因是它仍然满足“321原则" 改成多生产者多消费者的意义就是可以提高效率，比如单消费者在获取数据后需要将数据处理后才能重新获取下一个数据并处理，但是多消费者后在一个消费者处理数据的同时，另一个消费者就可以去获取数据，从而提高效率

总之，生产消费模型的功能还是挺多的，以上就是生产消费模型的原理和核心内容，感兴趣的可以再把我上面提到的那个传任务的方式试一下

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

4.1 环形队列

其实不管是以何种内存结构构件的生产消费模型，其核心都是一样的，这里我们主要是想结合POSIX信号量来创建一个生产消费模型，同时带大家再多看一个数据结构类型——环形队列搭建的生产消费模型

我们先来看一下环形队列是什么

学习过数据结构的应该对环形队列都有所了解，环形队列实际上就是队列首位相连的一种特殊形式，它最关键的地方就在于，我们如何判断其空满状态，我们是通过判断首位位置来做到的

在本处我们结合POSIX信号量来创建生产消费模型的过程中，实际上我们可以不用自己去判断队列中是否已满或为空，我们可以通过信号量来处理，因为我们之前讲过，信号量的本质就是一个计数器，它能够帮助我们统计我们所关注的资源的个数，在这里，生产者关注的就是还有多少个空位，我们称作空间资源，消费者关注的是有多少数据，我们称为数据资源

因为我们不需要去讨论队列为空或为满，所以其实我们可以用vector数组来替代环形队列

4.2 代码实现

代码实现与上面的阻塞队列的方法没啥大的区别，该注意的地方我在代码中注释出来了，这里就直接上代码了

Sem.hpp

#pragma once
#include<semaphore.h>

class Sem
{
public:
Sem(int value)
{
sem_init(&sem_,0,value);
}
void P()
{
sem_wait(&sem_);
}
void V()
{
sem_post(&sem_);
}
~Sem()
{
sem_destroy(&sem_);
}
private:
sem_t sem_;
};

RingQueue.hpp

#pragma once

#include"sem.hpp"
#include<iostream>
#include<pthread.h>
#include<vector>
using namespace std;

static int defaultcap=5;
template<class T>
class RingQueue
{
public:
RingQueue(int mp=defaultcap)
:max_cap(mp),rq(mp),p_index(0),c_index(0),
p_sem(mp),c_sem(0)
{
pthread_mutex_init(&p_mutex,nullptr);
pthread_mutex_init(&c_mutex,nullptr);
}
void Push(T& in)
{
p_sem.P();

pthread_mutex_lock(&p_mutex);
rq[p_index++]=in;
p_index%=max_cap;
pthread_mutex_unlock(&p_mutex);

c_sem.V();
}
T Pop()
{
c_sem.P();

pthread_mutex_lock(&c_mutex);
T out=rq[c_index++];
c_index%=max_cap;
pthread_mutex_unlock(&c_mutex);

p_sem.V();
return out;
}
~RingQueue()
{
pthread_mutex_destroy(&p_mutex);
pthread_mutex_destroy(&c_mutex);
}
private:
vector<T> rq; //用vector模拟环形队列
int max_cap; //队列的容量
int p_index; //生产者下标
int c_index; //消费者下标
Sem p_sem; //空间信号量（生产者所关注的信号量）
Sem c_sem; //数据信号量（消费者所关注的信号量）
pthread_mutex_t p_mutex;
pthread_mutex_t c_mutex;
};

Main.cc

#include"RingQueue.hpp"
#include<iostream>
#include<pthread.h>
#include<ctime>
#include<unistd.h>
using namespace std;

void *Productor(void *args)
{
//args是void*类型的，我们需要先转换
RingQueue<int> *rq=static_cast<RingQueue<int>*>(args);
//让我们的线程处于死循环的状态，即一直生产
while(true){
int in=rand()%10+1;

usleep(10); //模拟处理数据所需时间
rq->Push(in);
cout<<"Productor make a num: "<<in<<" threadid: "<<pthread_self()<<endl;
sleep(1); //让生产者在生产一个数据后睡眠一秒，便于观察打印结果
}
return nullptr;
}
void *Consumer(void *args)
{
RingQueue<int> *rq=static_cast<RingQueue<int>*>(args);
while(true){
int out=rq->Pop();
usleep(100); //模仿数据处理所需时间
cout<<"Consumer get a num: "<<out<<" threadid: "<<pthread_self()<<endl;
}
return nullptr;
}
int main()
{
srand(time(nullptr));
RingQueue<int> *rq=new RingQueue<int>();
pthread_t c,p;

//创建生产和消费两个线程，并且将bq作为参数传给两者
pthread_create(&p,nullptr,Productor,rq);
pthread_create(&c,nullptr,Consumer,rq);

//回收线程
pthread_join(c,nullptr);
pthread_join(p,nullptr);

delete rq;
return 0;
}

运行结果：

5. 总结

生产消费模型在外面平时的日常生活中还是比较常见的，它能够起到很好的解耦作用，提高系统整体效率和相应速度，同时，也能帮助我们更好的理解线程互斥与同步的问题，更加熟练的使用互斥锁、条件变量、POSIX信号量等，总之，学习完本篇生产消费模型，相信会对你对线程知识的掌握和如何提高系统整体效率的能力有更好的理解

本篇笔记：

感谢各位大佬观看，创作不易，还望各位大佬点赞支持！！！