Linux篇第十五篇——多线程（生产消费模型+POSIX信号量）
Posted 呆呆兽学编程
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Linux篇第十五篇——多线程（生产消费模型+POSIX信号量）相关的知识，希望对你有一定的参考价值。
⭐️ 本篇博客开始要继续给大家介绍线程同步和互斥的实际应用——生产者消费者模型，还要给大家介绍一种同步的机制——POSIX信号量
目录
🌏生产者消费者模型
概念： 生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过一个来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
生产消费者模型优点：
- 解耦：生产者和消费者是通过一个共享数据区域来进行通信。而不是直接进行通信，这样两个角色之间的依耐性就降低了（代码层面实现解耦），变成了角色与共享数据区域之间的弱耦合，一个逻辑出错不影响两一个逻辑，二者变得更独立。
- 支持并发：生产者负责生产数据，消费者负责拿数据。生产者生产完数据可以继续生产，大部分时间内是不需要等待消费者消费数据才继续生产。也就是说，在任一时刻，二者都是在正常处理任务的，进度都得以推进。
- 支持忙闲下不均：生产者生产了数据是放进容器中，消费者不必立即消费，可以慢慢地从容器中取数据。容器快要空了，消费者的消费速度就可以降下来，让生产者继续生产。
生产消费模型特征（简记321）：
- 3种关系： 生产者与生产者（互斥）、生产者与消费者（同步（主要）和互斥）和消费者与消费者（互斥）
- 两个角色： 生产者和消费者
- 一个交易场所： 容器、共享资源等
🌏基于BlockingQueue的生产者消费者模型
🌲介绍
在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。
阻塞队列的特点：
- 当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素
- 当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)
在这个模型中，生产者和消费者的交易场所就是阻塞队列。
🌲实现
🍯概述
因为交易场所是一个阻塞队列，所以，我封装了一个BlcokingQueue 的类，这里提供了放数据和取数据这样两个主要的方法。其中，有五个成员变量：
- 队列： 使用STL中的queue来实现
- 容量： 阻塞队列的容量，由用户给定，我们也可以提供一个默认的容量
- 互斥量： 为了实现生产者和消费者的同步，我们需要使用条件变量和互斥量来实现同步的操作
- 生产者唤醒和等待的条件变量： 当队列满了，生产者等待条件满足，应该挂起等待，等待消费者唤醒
- 消费者唤醒和等待的条件变量： 当队列为空，消费者等待条件满足，应该挂起等待，等待生产者唤醒
构造函数和析构函数：
- 构造函数做初始化和资源分配的操作，分配锁资源和条件变量
- 析构函数做清理资源的操作，对锁和条件变量进行销毁
代码整体框架：
template<class T>
class BlockQueue
public:
BlockQueue(int capacity = 5)
:_capacity(capacity)
pthread_mutex_init(&_lock, nullptr);
pthread_cond_init(&_c_cond, nullptr);
pthread_cond_init(&_p_cond, nullptr);
~BlockQueue()
pthread_mutex_destroy(&_lock);
pthread_cond_destroy(&_c_cond);
pthread_cond_destroy(&_p_cond);
private:
std::queue<T> _q;
size_t _capacity;// 队列最大容量
pthread_mutex_t _lock;// 锁
pthread_cond_t _c_cond;// 消费者被唤醒和挂起的条件变量
pthread_cond_t _p_cond;// 生产者被唤醒和挂起的条件变量
;
🍯基本方法的封装
我对阻塞队列的一些基本操作进行了封装，有以下几个处理动作（可以设置为私有方法）：
- 判断队列为空或为满
- 唤醒生产者和唤醒消费者
- 生产者挂起等待和消费者挂起等待
- 加锁和解锁
代码如下：
private:
bool IsFull()
return _q.size() == _capacity;
bool IsEmpty()
return _q.empty();
void ConsumerWait()
std::cout << "consumer wait...." << std::endl;
pthread_cond_wait(&_c_cond, &_lock);
void WakeUpConsumer()
std::cout << "wake up consumer...." << std::endl;
pthread_cond_broadcast(&_c_cond);
void ProductorWait()
std::cout << "productor wait...." << std::endl;
pthread_cond_wait(&_p_cond, &_lock);
void WakeUpProductor()
std::cout << "wake up productor...." << std::endl;
pthread_cond_broadcast(&_p_cond);
void LockQueue()
pthread_mutex_lock(&_lock);
void UnLockQueue()
pthread_mutex_unlock(&_lock);
🍯放数据和取数据
- 生产者进行相关操作前先上锁，队列如果为满就需要挂起等待。队列不为满就生成一个数据，然后需要把数据放入阻塞队列中，解开锁之后唤醒消费者，喊消费者进行消费。
- 消费者进行相关操作前先上锁，队列如果为空就需要挂起等待。队列不为空就需要从阻塞队列中取一个数据，然后解开锁之后唤醒消费者，喊生产者进行生产。
先看代码：
void ProductData(T data)
LockQueue();
while (IsFull())
// 让productor挂起等待
ProductorWait();
// 放数据
_q.push(data);
UnLockQueue();
// 唤醒consumer
WakeUpConsumer();
void ConsumeData(T& data)
LockQueue();
while (IsEmpty())
// 让consumer挂起等待
ConsumerWait();
// 取数据
data = _q.front();
_q.pop();
UnLockQueue();
// 唤醒productor
WakeUpProductor();
注意： 在临界资源判断唤醒条件是否就绪应该使用while循环检测，被唤醒的线程并不着急立即往下执行，而是再进行一次检测，判断当前唤醒条件是否真的就绪了。因为唤醒线程的这个函数调用可能会发生失败，且线程可能是在条件不满足的时候被唤醒，发生误判被伪唤醒。
🍯封装一个任务
我们可以实现一个任务类，生产者把这个任务放进阻塞队列中，消费者取出并进行处理。其中还有一个run的任务执行方法
实现如下：
class Task
public:
Task(int a = 0, int b = 0)
:_a(a)
,_b(b)
int Run()
return _a + _b;
int GetA()
return _a;
int GetB()
return _b;
private:
int _a;
int _b;
;
🍯单生产者和单消费者模型分析
为了更好地理解整个过程，我先用一个单生产者和单消费者模型来演示分析。
下面是生产者和消费者两个线程的执行代码：
BlockQueue<Task>* q;// 阻塞队列
void* Consumer(void* arg)
long id = (long)arg;
while (1)
// 消费（取）数据
Task t(0, 0);
q->ConsumeData(t);
std::cout << "consumer " << id << " consumes a task: " << t.GetA() << " + " << t.GetB() << " = " << t.Run() << std::endl;
sleep(1);// 后面可注释，调整速度
void* Productor(void* arg)
long id = (long)arg;
while (1)
// 生产（放）数据
int x = rand()%10 + 1;
int y = rand()%10 + 1;
Task t(x, y);
std::cout << "productor " << id << " produncs a task: " << x << " + " << y << " = ?" << std::endl;
q->ProductData(t);
sleep(1);// 后面可注释，调整速度
int main()
srand((size_t)time(nullptr));
// 创建一个交易场所
q = new BlockQueue<Task>;
pthread_t p, c;
pthread_create(&p, nullptr, Productor, (void*)(1));
pthread_create(&c, nullptr, Consumer, (void*)(1));
pthread_join(p, nullptr);
pthread_join(c, nullptr);
delete q;
return 0;
代码运行结果分三种情况进行分析：
- 消费者和生产者以相同的速度执行
可以看出的是，生产者生成完一个任务，消费者就处理了一个任务。接着生产者继续生产，消费者跟着处理，二者步调一致，一直并行的状态
2. 生产者快，消费者慢
生产者的速度慢，生产者生产一个任务立马唤醒消费者，消费者处理完一个数据，发现队列为空，然后挂起等待，接着生产者继续生产一个任务，然后唤醒消费者。可以看出的是，队列大部分时间是为空的，消费者大部分时间是处于挂起等待的，二者步调一直是一致的，且执行速度是随生产者的，也是并行的
3. 生产者慢，消费者快
生产者生产速度快，导致一上来生产者就把队列塞满了任务，接着唤醒消费者来消费，然后挂起等待。紧接着消费者处理完一个任务就唤醒生产者来生产，生产者生产了一个任务就喊消费者来消费，然后继续挂起。可以看出的是，在这种情况下，队列长时间是满的，生产者大部分时间是挂起等待的。生产者和消费者开始一小部分时间内步调是不一致的，生产者生产完，消费者才消费是串行的，但是过了一会，二者步调就变得一致了，且速度是随消费者的
注意： 这里我们还可以设置一个高水位线和低水位线，队列中任务量达到高水位线时就喊消费者来消费，任务量达到低水位线时，就喊生产者赶紧来生产。
🍯多生产者和多消费者
多生产者和多消费者不做详细分析，我会把主体内容介绍。
做到几点：
- 生产者之间需要互斥，也就是生产者和生产者之间需要组内竞争一把锁，消费者也是如此
- 生产者和消费者之间用互斥量和条件变量做到同步和互斥（上面就做到了）
代码如下：
#define P_COUNT 3
#define C_COUNT 3
BlockQueue<Task>* q;
pthread_mutex_t c_lock;// 消费者的锁
pthread_mutex_t p_lock;// 生产者的锁
void* Consumer(void* arg)
long id = (long)arg;
while (1)
pthread_mutex_lock(&c_lock);
// 消费（取）数据
Task t(0, 0);
q->ConsumeData(t);
std::cout << "consumer " << id << " consumes a task: " << t.GetA() << " + " << t.GetB() << " = " << t.Run() << std::endl;
pthread_mutex_unlock(&c_lock);
sleep(1);
void* Productor(void* arg)
long id = (long)arg;
while (1)
pthread_mutex_lock(&p_lock);
// 生产（放）数据
int x = rand()%10 + 1;
int y = rand()%10 + 1;
Task t(x, y);
std::cout << "productor " << id << " produncs a task: " << x << " + " << y << " = ?" << std::endl;
q->ProductData(t);
pthread_mutex_unlock(&p_lock);
sleep(1);
int main()
srand((size_t)time(nullptr));
pthread_mutex_init(&c_lock, nullptr);
pthread_mutex_init(&p_lock, nullptr);
// 创建一个交易场所
q = new BlockQueue<Task>;
pthread_t p[P_COUNT];
pthread_t c[C_COUNT];
for (long i = 0; i < P_COUNT; ++i)
pthread_create(p+i, nullptr, Productor, (void*)(i+1));
for (long i = 0; i < C_COUNT; ++i)
pthread_create(c+i, nullptr, Consumer, (void*)(i+1));
for (int i = 0; i < C_COUNT; ++i)
pthread_join(c[i], nullptr);
for (int i = 0; i < P_COUNT; ++i)
pthread_join(p[i], nullptr);
pthread_mutex_destroy(&c_lock);
pthread_mutex_destroy(&p_lock);
delete q;
return 0;
注意：
- 生产者之间需要一个互斥量，消费者之间也需要一个互斥量
- 生产者和消费者的个数可以自己调整宏变量
- 主体代码并没有什么大的改变
🌏POSIX信号量
🌲介绍
POSIX信号量： 该信号量允许进程和线程同步对共享资源的访问。同时也可以用于实现线程间同步。
总结几点：
- 是什么？ 信号量本质是一个计数器，描述临界资源的有效个数。申请一个资源就对信号量减1（P操作），释放一个资源就对信号量加1（V操作）
- 为什么？ 临界资源可以看成很多份，互相不冲突且高效
- 怎么用？ 可以使用信号量的相关接口，来申请信号量和释放信号量（下面详细介绍）
🌲相关接口介绍
下面要介绍的POSIX信号量相关接口都是在semaphore.h的头文件中。信号量是一个类型为sem_t的变量
初始化信号量：
函数原型：
int sem_init(sem_t *sem, int pshared, unsigned int value);参数：
- sem：信号量
- pshared：0表示线程间共享，非零表示进程间共享
- value：信号量初始值
返回值： 成功返回0，失败返回-1
销毁信号量：
函数原型：
int sem_destroy(sem_t *sem);参数：
- sem：信号量
返回值： 成功返回0，失败返回-1
等待信号量：
函数原型：
int sem_wait(sem_t *sem);功能： 等待信号量，会将信号量的值减1
参数：
- sem：信号量
返回值： 成功返回0，失败返回-1
发布信号量：
函数原型：
int sem_post(sem_t *sem);功能： 发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1
参数：
- sem：信号量
返回值： 成功返回0，失败返回-1
信号量的简单使用：
#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
using namespace std;
sem_t sem;
void* run1(void* arg)
while (1)
sem_wait(&sem);
cout << "run1 is running..." << endl;
sem_post(&sem);
sleep(1);
void* run2(void* arg)
while (1)
sem_wait(&sem);
cout << "run2 is running..." << endl;
sem_post(&sem);
sleep(1);
int main()
sem_init(&sem, 0, 1);
pthread_t t1, t2;
pthread_create(&t1, nullptr, run1, nullptr);
pthread_create(&t2, nullptr, run2, nullptr);
sem_destroy(&sem);
pthread_join(t1, nullptr);
pthread_join(t2, nullptr);
return 0;
代码运行结果如下： 通过信号量可以做到线程同步
说明： 二元信号量（value=1，一个资源）等价于互斥锁
🌏基于环形队列的生产消费模型
🌲环形队列介绍
环形队列： 环形队列和普通队列的区别就是，这种队列是一种环形的结构，有一个头指针和一个尾指针维护环中一小段队列。（如下图）
环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置，作为满的状态。
由于环形队列采用数组模拟，用模运算来模拟环状特性，如下：
因为信号量就是一个计数器，所以我们可以通过信号量来实现多线程间的同步过程。
🌲实现
🍯概述
一个交易场所： 循环队列
两个角色：
- 生产者：需要申请空间资源（P操作），然后释放数据资源（V操作）
- 消费者：需要申请数据资源（P操作），然后释放空间资源（V操作）
三种关系： 生产者与生产者（互斥）、生产者与消费者（同步（主要）和互斥）和消费者与消费者（互斥）
几个变量成员：
- 队列：数组模拟
- 容量：由用户给定
- 空间资源信号量：队列的容量大小
- 数据资源信号量：开始为0
- 生产者的下标位置：开始为0
- 消费者的下标位置：开始为0
代码如下：
template<class T>
class RingQueue
public:
RingQueue(int capacity = 5)
:_capacity(capacity)
,_rq(capacity)
,_c_index(0)
,_p_index(0)
sem_init(&_blank_sem, 0以上是关于Linux篇第十五篇——多线程（生产消费模型+POSIX信号量）的主要内容，如果未能解决你的问题，请参考以下文章
Python学习笔记——进阶篇第九周———线程进程协程篇（队列Queue和生产者消费者模型）