【Linux系统】生产者消费者模型：基于环形队列（信号量机制）

理论层面 1、环形队列的特性认识 环形队列采用数组模拟，用模运算来模拟环状特性

---

---

环形结构起始状态和结束状态都是⼀样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留⼀个空的位置，作为满的状态

---

---

2、环形队列的生产消费模型

对于计算机而言，本质上只有两种资源：空间和数据。

环形队列的生产消费模型可以理解为一种固定存放数据的空间机制。在这个模型中，生产者申请空位置写入数据，而消费者从非空的位置获取数据。

---

---

3、环形队列的特性

互斥 当环形队列为“空”时，只有生产者能够进入并写入数据；当环形队列为“满”时，只有消费者能够进入并读取数据。这种机制保证了生产者和消费者之间的互斥性。

同步 当环形队列为“空”时，必须让生产者先进行生产操作；当环形队列为“满”时，必须让消费者先进行消费操作。这种机制体现了生产者和消费者之间的同步关系。

4、并发与追逐游戏

在环形队列不为空也不为满的情况下，生产者和消费者的关系就像一场在圆桌上的追逐游戏：

生产者不停地绕着圆桌，在每个空盘子上放置苹果。消费者则紧跟其后，不停地从盘子中拿走苹果。

在这种情况下，生产者和消费者的操作是并发的，彼此独立且互不影响。

5、特殊情况 当消费者追上生产者时，说明消费者的消费能力较强，很快将圆桌上的苹果全部拿完，此时环形队列处于“空”的状态。当生产者追上消费者时，说明生产者的生产能力较强，很快将圆桌上的苹果全部放满，此时环形队列处于“满”的状态。

由于环形队列的“空”和“满”状态会触发互斥与同步机制，因此消费者绝对不可能超越生产者。

6、资源需求与信号量设计 对于生产者来说，需要的是空间资源，因为生产者需要向空的位置写入数据。对于消费者来说，需要的是物品资源，因为消费者需要从非空位置读取数据。

基于此，可以设计两个信号量：

空间信号量：用于表示可用的空闲空间。物品信号量：用于表示已有的数据资源。

综上所述，生产者和消费者在同一位置时体现同步和互斥，而在不同位置时体现并发关系。

代码层面 1、生产者与信号量的操作逻辑

生产者对空间的 P 操作 生产者需要占用一个空闲空间来写入数据。因此，在生产之前，它会对空间信号量进行 P 操作（申请资源）。这表示生产者正在尝试获取一个可用的空间。

生产者对数据的 V 操作 当生产者完成数据写入后，它并不会释放自己占用的空间（即不会对空间信号量进行 V 操作），而是通过 V 操作增加数据信号量的值。这是因为生产者的核心任务是生成数据，并将数据提供给消费者使用。

空间信号量：表示可用的空闲空间数量。数据信号量：表示已生成的数据数量。

因此，生产者的操作逻辑可以总结为：

对空间信号量 P 操作：申请一个空闲空间。对数据信号量 V 操作：通知消费者有新的数据可以消费。

---

2、信号量的本质与作用

信号量的作用并不是直接控制生产者或消费者的具体位置，而是通过数值的变化来控制生产和消费的次数。换句话说，生产者和消费者只需要检查各自信号量的值，就能判断是否可以进入临界区执行操作。

空间信号量：控制生产者的生产次数。当空间信号量的值大于 0 时，生产者可以进入临界区；否则必须等待。数据信号量：控制消费者的消费次数。当数据信号量的值大于 0 时，消费者可以进入临界区；否则必须等待。

通过这种机制，信号量实现了以下功能：

同步：确保生产者和消费者按照正确的顺序操作。例如，当队列为空时，生产者优先生产；当队列为满时，消费者优先消费。互斥：避免生产者和消费者同时访问同一个位置，从而保护共享资源的完整性。

---

3、程序雏形：单生产者单消费者 RingBuffer.hpp #ifndef _RING_BUFFER_HPP_ #define _RING_BUFFER_HPP_ #include <iostream> #include <queue> #include <pthread.h> #include <vector> #include "Cond.hpp" #include "Mutex.hpp" #include "Sem.hpp" using namespace CondModule; using namespace MutexModule; using namespace SemModule; // 使用封装的条件变量和互斥锁 namespace RingBufferModule { // 数据 int num = 10; template <typename T> class RingBuffer { public: RingBuffer(size_t cap = 10) : _cap(cap), _ring(cap), _producer_index(0), _consumer_index(0), _space(_cap), _data(0) { } ~RingBuffer() { } void Equeue(const T &in) { // 生产者生产数据, 生产者争夺锁用于争夺信号量的使用权 _space.P(); _ring[_producer_index] = in; _producer_index++; _producer_index %= _cap; _data.V(); } void Pop(T *out) { // 消费者消费数据, 消费者争夺锁用于争夺信号量的使用权 _data.P(); *out = _ring[_consumer_index]; _consumer_index++; _consumer_index %= _cap; _space.V(); } private: std::vector<T> _ring; int _cap; // 容量 int _producer_index; // 生产者下标 int _consumer_index; // 消费者下标 Sem _space; Sem _data; }; } #endif Main.cc #include "RingBuffer.hpp" #include "Task.hpp" #include "Sem.hpp" #include <unistd.h> #include <stdlib.h> #include <time.h> #include <functional> using namespace RingBufferModule; using namespace SemModule; void *Producer(void *args) { RingBuffer<int>* ring_buffer = static_cast<RingBuffer<int>*>(args); int data = 0; while (true) { ring_buffer->Equeue(data); std::cout << "生产者生产了一个数据: " << data++ << '\n'; } } void *Consumer(void *args) { RingBuffer<int>* ring_buffer = static_cast<RingBuffer<int>*>(args); while (true) { sleep(1); int data; ring_buffer->Pop(&data); std::cout << "消费者消费了一个数据: " << data << '\n'; } } int main() { srand(time(0) ^ getpid()); RingBuffer<int> ring_buffer; // 创建生产者线程 pthread_t _tid1; pthread_create(&_tid1, nullptr, Producer, &ring_buffer); // 创建消费者线程 pthread_t _tid10; pthread_create(&_tid10, nullptr, Consumer, &ring_buffer); // 回收线程 pthread_join(_tid1, nullptr); pthread_join(_tid10, nullptr); return 0; } Sem.hpp ：对信号量的封装 #pragma once #include <iostream> #include <semaphore.h> namespace SemModule { class Sem { public: Sem(int value) : _init_val(value) { sem_init(&_sem, 0, _init_val); } ~Sem() { sem_destroy(&_sem); } // P操作 void P() { sem_wait(&_sem); } // V操作 void V() { sem_post(&_sem); } private: sem_t _sem; int _init_val; }; }

运行结果如下，可以看到消费者正按照一定顺序逐个消费数据

以下是代码设计的相关讨论：

为什么信号量代码中没有判断是否为空或为满？ 1. 锁的作用

在使用互斥锁的代码中，我们需要手动判断队列是否为空或为满：

锁的作用只是控制对临界区的访问权限。锁并不知道临界区内资源（数据）的具体状态。因此，我们必须通过额外的逻辑来判断队列的状态（空或满），以决定是否允许生产或消费。锁只是一个看门的，至于门里的数据什么情况，和锁无关 2. 信号量的作用

信号量本身的设计已经包含了对资源数量的记录和管理：

空间信号量：表示队列中有多少个空闲位置。数据信号量：表示队列中有多少个已生成的数据。

当信号量的操作成功时（P 操作通过），说明当前确实有可用资源，无需再单独判断队列是否为空或为满。信号量机制直接将资源的管理内化到其操作逻辑中，简化了代码。

总结来说：

锁需要判断空满是因为它只负责访问控制，不了解资源状态。信号量不需要判断空满是因为它的初始值和操作本身就反映了资源的状态。

---

单生产者单消费者 vs 多生产者多消费者 1、单生产者单消费者

在这种情况下，主要需要解决的是生产者和消费者之间的同步与互斥关系：

同步：确保生产者和消费者按照正确的顺序操作（生产者先生产，消费者后消费）。互斥：避免生产者和消费者同时访问同一个位置。

信号量机制可以很好地解决这些问题：

使用空间信号量控制生产者的生产次数。使用数据信号量控制消费者的消费次数。 2、多生产者多消费者

在这种情况下，除了生产者和消费者之间的同步与互斥关系外，还需要解决以下问题：

生产者之间的互斥 多个生产者可能同时尝试写入队列中的某个位置，因此需要保证生产者之间的互斥。 消费者之间的互斥 多个消费者可能同时尝试读取队列中的某个位置，因此需要保证消费者之间的互斥。

---

如何解决多生产者多消费者的互斥问题：加锁 1. 是否可以用一把锁？

理论上可以用一把锁来解决所有问题，但这会导致 生产和消费被完全串行化：

生产者和消费者都需要等待对方释放锁才能继续操作。这种方式会 严重影响程序的并行性，失去了信号量机制的优势。

此外， 若只定义了一把锁可能会导致死锁问题：

生产者先获取锁，然后阻塞在对空间的 P 操作上，而某个消费者需要锁进来入临界区代码进行空间的 V 操作，生产者只有获取该消费者给的空间才能继续执行代码，造成了死锁。 2. 应该用几把锁？

建议使用两把锁：

生产者锁：用于保证多个生产者之间的互斥。消费者锁：用于保证多个消费者之间的互斥。

这种设计既能保证生产者和消费者之间的同步，又能保证生产者之间和消费者之间的互斥，同时不会完全串行化生产和消费操作。

3. 加锁的目的 加锁的目的：每次进行生产或消费时，生产者或消费者需要争抢得出一个代表，进入临界区进行生产或消费。有了锁之后，多生产者和多消费者的情况本质上变成了单生产者和单消费者的情况，因为读写位置只有一个！

---

锁与信号量的结合使用

我想了想，本来想着用锁锁住生产者和消费者对队列下标的使用权，但经过思考后发现：

如果已经通过信号量成功申请到了资源（即进入了临界区），说明当前确实有可用的空间或数据，无需再争抢下标。因此，锁的作用可以转换为争夺信号量的使用权：多个线程同时抢到了一个信号量，我们需要选出一个代表进行真正的生产或消费工作，这就是用锁控制多线程对信号量的使用权。

---

锁加在信号量之前还是之后？ 后锁 因为多个线程可以对同一个信号量进行 P 操作，所以会出现多线程同时拥有生产或消费权力的情况。我们设计的锁本质上是为了控制对空间或物资的使用权，而空间或物资的使用权本质上就是对相应信号量的使用权。因此，可以将锁加在信号量的 P 操作之后，表示每个线程对信号量进行 P 操作后，还需要获取锁，才能真正进入临界区进行生产或消费。这样，后锁也就保证了互斥关系，保证每次只能有一个生产者进入临界区或只能有一个消费者进入临界区 void Equeue(const T& in) { // 生产者生产数据, 生产者争夺锁用于争夺信号量的使用权 _space.P(); _p_mux.lock(); //...临界区 _p_mux.unlock(); _data.V(); } void Pop(T* out) { // 消费者消费数据, 消费者争夺锁用于争夺信号量的使用权 _data.P(); _c_mux.lock(); //...临界区 _c_mux.unlock(); _space.V(); } 先锁

所有线程必须先获取锁，才能进一步申请信号量，进而进入临界区生产或消费

这个过程，也保证了多生产者之间和多消费者之间的互斥关系

先锁好还是后锁好？

无论先锁还是后锁，都能保证互斥关系，但后锁更优：对于同一个信号量，多线程可以并行的申请！若前锁，则导致获取锁和获取信号量都是串行的，影响效率；若后锁，则可以显然信号量被所有线程并行的申请，无需串行的等待对方，如此提高了一定效率

因此一般都是先信号量，再加锁!

void Equeue(const T& in) { // 生产者生产数据, 生产者争夺锁用于争夺信号量的使用权 _space.P(); _p_mux.lock(); //... _p_mux.unlock(); _data.V(); } void Pop(T* out) { // 消费者消费数据, 消费者争夺锁用于争夺信号量的使用权 _data.P(); _c_mux.lock(); //... _c_mux.unlock(); _space.V(); }

用上 LockGuard 让代码更加优雅一点！

void Equeue(const T &in) { // 生产者生产数据, 生产者争夺锁用于争夺信号量的使用权 _space.P(); { LockGuard p_lock(_p_mux); _ring[_producer_index] = in; _producer_index++; _producer_index %= _cap; } _data.V(); } void Pop(T *out) { // 消费者消费数据, 消费者争夺锁用于争夺信号量的使用权 _data.P(); { LockGuard c_lock(_c_mux); *out = _ring[_consumer_index]; _consumer_index++; _consumer_index %= _cap; } _space.V(); }

因此最终代码如下：

完整版代码 #ifndef _RING_BUFFER_HPP_ #define _RING_BUFFER_HPP_ #include <iostream> #include <queue> #include <pthread.h> #include <vector> #include "Cond.hpp" #include "Mutex.hpp" #include "Sem.hpp" using namespace CondModule; using namespace MutexModule; using namespace SemModule; // 使用封装的条件变量和互斥锁 namespace RingBufferModule { // 数据 int num = 10; template <typename T> class RingBuffer { public: RingBuffer(size_t cap = 10) : _cap(cap), _ring(cap), _producer_index(0), _consumer_index(0), _space(_cap), _data(0) { } ~RingBuffer() { } void Equeue(const T &in) { // 生产者生产数据, 生产者争夺锁用于争夺信号量的使用权 _space.P(); { LockGuard p_lock(_p_mux); _ring[_producer_index] = in; _producer_index++; _producer_index %= _cap; } _data.V(); } void Pop(T *out) { // 消费者消费数据, 消费者争夺锁用于争夺信号量的使用权 _data.P(); { LockGuard c_lock(_c_mux); *out = _ring[_consumer_index]; _consumer_index++; _consumer_index %= _cap; } _space.V(); } private: std::vector<T> _ring; int _cap; // 容量 int _producer_index; // 生产者下标 int _consumer_index; // 消费者下标 // 锁 Mutex _p_mux; // 生产者锁 Mutex _c_mux; // 消费者锁 Sem _space; Sem _data; }; } #endif