【Linux 22】生产者消费者模型

🌈 一、生产者消费者模型

⭐ 1. 生产者消费者模型的概念

• 生产者消费者模式通过一个容器来解决生产者和消费者之间的强耦合问题。
• 生产者和消费者之间不会直接通讯，而是通过这个容器来进行通讯。
• 生产者生产完数据之后不用等消费者处理，而是直接将生产出的数据放到这个容器，消费者也不找生产者要数据，而是直接从这个容器里取。
  • 简单理解成 供应商 → 超市 → 消费者 就行。
• 一般是使用阻塞队列 / 环形队列作为生产者和消费者之间通讯的容器，这个容器相当于一个缓冲区，平衡了生产者和消费者的处理能力。
• 这个阻塞队列就是专门用来给生产者和消费者解耦的。

⭐ 2. 生产者消费者模型的特点

1. 生产者与消费者之间的 321 原则

• 3 种关系：生产者和生产者 (互斥关系)、消费者和消费者 (互斥关系)、生产者和消费者 (互斥 / 同步关系) 。
• 2 种角色：生产者和消费者 (通常由 进程 / 线程 充当这两种角色) 。
• 1 个场所：生产者和消费者交易的场所，通常指的是内存中的一段缓冲区，也能自己通过某种方式组织起来。

2. 为什么生产者和消费者、消费者和消费者、生产者和生产者之间会存在互斥关系

• 生产者和消费者：生产者在放数据时不知道自己有没有放完数据，消费者在拿数据时也不知道自己有没有拿到数据，如果不互斥，在访问同一个位置时就可能产生数据不一致的问题，因此需要竞争互斥锁。

• 生产者和生产者：在访问同一个位置时，可能生产者 A、B 都认为这个位置没东西，就会都往这里放数据，会导致先放的数据会被后放的数据覆盖。

• 消费者和消费者：消费者 A、B 同时访问一个位置，都认为这个位置有数据，但 B 先把东西拿走了，A 不知道数据已经被取走了，继续去拿数据，导致拿了个寂寞。

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

• 如果让生产者一直生产数据，当容器被数据塞满后，生产者再进行生产就会导致生产失败。
• 如果让消费者一直拿取数据，当容器数据被搬空后，消费者再进行消费就会导致消费失败。
• 虽然，一直生产或一直消费不会导致数据不一致的问题，但却会引发饥饿问题。
• 应该让生产者和消费者之间在访问容器时具有一定的顺序性。
  • 当容器的管理者发现容器内数据量下降到标准线之下时，就让消费者停止消费，通知生产者生产数据。
  • 当容器中的数据量被填充到标准线之上时，就让生产者停止生产，通知消费者过来消费。

4. 互斥保证数据的准确性，同步将多线程协同起来，两者并不冲突

⭐ 3. 生产者消费者模型的优点

1. 解耦

• 生产者只负责生产数据，消费者只负责消费数据，两者之间互不影响。

• 从代码层面看，生产者线程和消费者线程的代码并不直接互相调用，两者的代码在发生变化不会对对方造成影响。

2. 支持并发

• 在消费者从缓冲区拿取数据后的处理数据期间，生产者可以同时进行生产对缓冲区添加数据。
• 如果没有缓冲区，消费者得直接去找生产者要数据，就必须等待生产真产生数据，同理，生产者也需要等待消费者消费数据。

3. 支持忙闲不均

• 在缓冲区未满时，生产者和消费者互不影响，不会产生占用 CPU 时间片的问题；
• 在缓冲区已满时，生产者不再生产数据，在缓冲区空时，消费者不再消费数据，使得两者总体处于一种动态平衡的状态。

🌈 二、基于阻塞队列的生产消费模型

⭐ 1. 阻塞队列概念

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

阻塞队列和一般队列的区别

• 当阻塞队列为空时，消费者线程从阻塞队列中获取元素的操作会被阻塞，直到阻塞队列中被生产者线程放入元素为止；
• 当阻塞队列为满时，生产者线程往阻塞队列中存放数据的操作会被阻塞，直到阻塞队列因为被消费者线程拿走元素而出现空位为止。

⭐ 2. 模拟实现基于阻塞队列的生产消费模型

• 根据 C++ 中的 queue 容器实现一个阻塞队列。
• 为了方便演示，实现的是一个单生产者、单消费者的模型。

• 当生产者线程把阻塞队列填满时，通知消费者线程消费；当消费者线程把阻塞队列搬空时，通知生产者线程生产。
  • 也可以不这么极端，可以设置一个标准水位线，当阻塞队列中的数据量 < 水位线时，让生产者生产数据，> 水位线时，让消费者消费数据。
  • 这里就实现 空 / 满 这种极端做法实现。
• 注：由生产者线程通知消费者线程消费，由消费者线程通知生产者线程生产。
  • 当消费者线程发现阻塞队列空了之后，就要通知生产者线程生产数据，然后消费者去指定条件变量处等待。
  • 当生产者线程发现阻塞队列满了之后，就要通知消费者线程消费数据，然后生产者去指定条件变量处等待。

#include <queue>
#include <iostream>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>using std::cout;
using std::endl;
using std::queue;const int num = 6;          // 设阻塞队列的容量为 6template<typename T>
class block_queue
{
private:queue<T> q;           // 用队列实现阻塞队列, 阻塞队列属于临界资源int cap;                // 阻塞队列的容量pthread_mutex_t lock;   // 互斥锁, 用于保护使用阻塞队列的临界区pthread_cond_t full;    // 生产者线程用的条件变量, 当阻塞队列满时去这里等待pthread_cond_t empty;   // 消费者线程用的条件变量, 当阻塞队列空时去这里等待private:// 上锁void lock_queue(){pthread_mutex_lock(&lock);}// 解锁void unlock_queue(){pthread_mutex_unlock(&lock);}// 让生产者线程去 full 条件变量处等待void product_wait(){pthread_cond_wait(&full, &lock);}// 让消费者线程去 empty 条件变量处等待void consume_wait(){pthread_cond_wait(&empty, &lock);}// 唤醒在 full 条件变量队列等待的的队头的生产者线程void notify_product(){pthread_cond_signal(&full);}// 唤醒在 empty 条件变量队列等待的的队头的消费者线程void notify_consume(){pthread_cond_signal(&empty);}// 判断当前阻塞队列是否为空bool is_empty(){return 0 == q.size();}// 判断当前阻塞队列是否为满bool is_full(){return q.size() == cap;}public:// 构造函数block_queue(int _cap = num) : cap(_cap){pthread_mutex_init(&lock, nullptr);	// 初始化互斥锁pthread_cond_init(&full, nullptr);	// 初始化生产者线程用的条件变量pthread_cond_init(&empty, nullptr);	// 初始化消费者线程用的条件变量}// 生产者线程往阻塞队列中放数据void push_data(const T &data){lock_queue();		 	// 下面的代码要访问临界资源 (阻塞队列) 了, 是临界区, 上锁while (is_full())       // 阻塞队列满了{notify_consume();   // 通知消费者线程消费数据cout << "阻塞队列满了, 通知消费者消费数据, 让生产者停止生产" << endl;product_wait();     // 让生产者线程等待，不要再生产了}q.push(data);           // 阻塞队列没满时，生产者线程会一直往里面放数据unlock_queue();			// 访问完临界区了, 解锁}// 消费者线程从阻塞队列中拿数据void pop_data(T *data){lock_queue();			// 下面的代码要访问临界资源 (阻塞队列) 了, 是临界区, 上锁while (is_empty())  	// 阻塞队列空了{notify_product();	// 通知生产者线程生产数据cout << "阻塞队列空了, 通知生产者生产数据, 让消费者停止消费" << endl;consume_wait();		// 让消费者线程等待，不要再拿取数据了}*data = q.front();		// 阻塞队列没空时，消费者线程会从里面拿取数据q.pop();unlock_queue();			// 访问完临界区了, 解锁}// 析构函数~block_queue(){pthread_mutex_destroy(&lock);   // 销毁保护阻塞队列的互斥锁pthread_cond_destroy(&full);    // 销毁生产者线程用的条件变量pthread_cond_destroy(&empty);   // 销毁消费者线程用的条件变量}
};// 消费者线程调用的函数
void *consumer(void *arg)
{int data;block_queue<int> *bqp = (block_queue<int> *)arg;while (true){bqp->pop_data(&data);    // 消费者线程用 data 从阻塞队列获取数据cout << "消费者拿取数据完毕, 拿取的数据是: " << data << endl;sleep(1);}
}// 生产者线程调用的函数
void *producter(void *arg)
{// 生产者线程生产的数据是一些随机数block_queue<int> *bqp = (block_queue<int> *)arg;srand((unsigned long)time(NULL));       // 随机数种子while (true){int data = rand() % 1024;           // 随机数bqp->push_data(data);               // 将生产者生产的随机数放入阻塞队列cout << "生产者生产数据完毕, 生产的数据是: " << data << endl;sleep(1);}
}int main()
{block_queue<int> bq; 	// 阻塞队列对象pthread_t cid;  		// 消费者线程的 idpthread_t pid;  		// 生产者线程的 id// 创建生产者和消费者线程, 让它们去执行各自的函数pthread_create(&cid, nullptr, consumer, (void *)&bq);    // 将阻塞队列对象作为参数传递给消费者线程调用的函数pthread_create(&pid, nullptr, producter, (void *)&bq);   // 将阻塞队列对象作为参数传递给生产者线程调用的函数pthread_join(cid, nullptr); // 主线程等待消费者线程pthread_join(pid, nullptr); // 主线程等待生产者线程return 0;
}

🌈 三、POSIX 信号量

⭐ 1. POSIX 信号量概念

• POSIX 信号量的作用和 SystemV 信号量相同，都是用于同步操作，从而达到无冲突的访问资源的目的；但 POSIX 信号量可以用于线程间同步。

🌙 1.1 信号量的本质

• POSIX 信号量本质上是一个计数器，用于描述临界资源数量的计数器，能够更细粒度的对临界资源进行管理。

  • 假设当前有一个可以容纳 100 个人的博物馆，信号量的初始值就是 100，表示博物馆可提供的资源数。
  • 每当进去一个人时，信号量的值就减 1；每当出去一个人时，信号量的值就加 1。
  • 当信号量的值为 0 时，说明博物馆已经装不下更多人了，让后面的游客 (线程) 一直等到信号量不为 0。

• 执行流在进入临界区之前，都应该申请信号量，申请成功后才具备操作临界资源的权限，当操作完毕后也应该释放信号量。

🌙 1.2 信号量的 PV 操作

• P 操作：将申请信号量称为 P 操作，申请信号量本质就是申请对临界资源的使用权限，当申请成功时会让信号量的值 - 1。因此 P 操作的本质就是让信号量这个计数器的值 - 1。
• V 操作：将释放信号量成为 V 操作，释放信号量本质就是归还对临界资源的使用权限，当释放成功时会让信号量的值 + 1。因此 V 操作的本质就是让信号量这个计数器的值 + 1。

信号量的 PV 操作必须是原子的

• 和争锁一样，多执行流之间为了访问临界资源也会竞争信号量，因此信号量也会被多执行流同时访问，即信号量本身就是个临界资源。
• 但信号量本质上是为了保护临界资源，不可能说再弄个信号量或锁去保护信号量，因此信号量的 PV 操作必须是原子的。

⭐ 2. POSIX 信号量函数

• 信号量的数据类型是 sem_t，可以使用该类型定义信号量。
• 信号量函数的返回值：调用函数成功时返回 0，失败时返回 - 1。
• 在使用信号量函数之前，应该先使用 #include <semaphore.h> 引入库文件。

🌙 2.1 初始化信号量

#include <semaphore.h>int sem_init(sem_t *sem, 		/* sem 表示需要初始化的信号量 */int pshared, 		/* 设置 sem 的共享方式，为 0 在线程间共享，非 0 在进程间共享 */unsigned int value); /* 信号量 (计数器) 的初始值 */

🌙 2.2 销毁信号量

#include <semaphore.h>int sem_destroy(sem_t *sem);	// sem 表示要销毁的信号量

🌙 2.3 申请信号量 (等待信号量)

• 申请信号量 (等待信号量) 就是 PV 操作中的 P 操作。
  • 注：申请信号量的 P 操作应该在线程争锁之前进行，只有在确定有资源的情况下才让线程去争锁。

#include <semaphore.h>int sem_wait(sem_t *sem);	// sem 表示线程需要申请的信号量

• 调用该函数时，如果申请信号量成功，则让信号量的值 - 1；
• 如果申请信号量失败，则将调用该函数的线程在 sem 信号量的等待队列处挂起等待。

🌙 2.4 释放信号量 (发布信号量)

• 释放信号量 (发布信号量) 就是 PV 操作中的 V 操作。
  • 注：释放信号量的 V 操作应该在线程解锁之后进行。

#include <semaphore.h>int sem_post(sem_t *sem);	// sem 表示线程需要释放的信号量

• 调用该函数时，如果释放信号量成功，则让信号量的值 + 1。

🌈 四、基于环形队列的生产消费模型

⭐ 1. 环形队列的概念

• 环形队列与阻塞队列最大的不同就是，环形队列能够构成一个环。
• 在环形队列中，生产者和消费者一开始可以指向同一个位置，启动之后，让生产者先生产数据，消费者跟在后面消费数据。

🌙 1.1 空间资源和数据资源

• 并不一定说，只有数据才是资源，角色的不同，也会导致对资源的认识不同。

• 对于生产者来说，它关心的是缓冲区的空间资源，而对于消费者来说，它关心的是缓冲区中的数据资源。

• 缓冲区中只要有空间，生产者就能干活；同理，缓冲区中只要有数据，消费者就能干活。

⭐ 2. 环形队列的规则

🌙 2.1 生产者和消费者不能访问同一个位置

• 如果生产者和消费者访问的是环形队列中相同的位置，就可能会出现数据不一致的问题。
• 消费者想拿旧数据，生产者新生产出的数据可能会将该位置旧有的数据给覆盖掉，但消费者又不知道自己拿到的是新的数据。

🌙 2.2 生产者不能快消费者一圈以上

• 如果生产者线程跑的太快了，绕一圈回来撞上在后面拿数据的消费者线程，如果生产者此时还不停下来，就会覆盖掉之前的数据。
• 这条规则本质上就是生产者和消费者不能访问同一个位置。

🌙 2.3 消费者不能超过生产者

• 消费者已经将环形队列中的数据消费完了，此时消费者已经追上了生产者。
• 如果消费者想超过生产者继续往前走，前面就不会有数据可供消费者消费了。
• 就算前面有数据，消费者拿到的也是之前用过的旧数据，并不是所需的新数据。

⭐ 3. 生产者消费者并发访问环形队列的场景

1. 以下情况会导致生产者和消费者访问同一个位置，不能实现并发访问

1. 环形队列为空：消费者消费了一圈追上了生产者，生产者和消费者访问的是同一个位置，此时消费者应该停止消费，让生产者去生产。

2. 环形队列为满：生产者生产力一圈追上了消费者，生产者和消费者访问的是同一个位置，此时生产者应该停止生产，让消费者去消费。

2. 生产者和消费者不访问同一位置时，可以实现并发访问

• 生产者在前面跑，消费者在后面追，两者之间的距离小于一圈。
• 这样可以让生产者一直能够获取空间资源，让消费者一直能够获取数据资源。

⭐ 4. 模拟实现基于环形队列的生产消费模型

• 想要判断环形队列是否为 空 / 满，可以通过计数器。也可以预留一个位置作为满的状态。
• 在了解了信号量之后，就可以使用信号量作为环形队列中资源数量的计数器。
  • 可以定义两个信号量，分别是给生产者用的空间资源信号量，以及给消费者用的数据资源信号量。
• 当前要实现的是让生产者生产者先生产一个随机数，然后让消费者消费这个随机数，生产者与消费者之间就差一个身位。

#include <vector>
#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>using std::cout;
using std::endl;
using std::vector;const int default_cap = 4;              // 环形队列的默认大小// 定义并初始化一个生产者之间的互斥锁
pthread_mutex_t p_mutex = PTHREAD_MUTEX_INITIALIZER;// 定义并初始化一个消费者之间的互斥锁
pthread_mutex_t c_mutex = PTHREAD_MUTEX_INITIALIZER;template<typename T>
class ring_queue
{
private:vector<T> q;                        // 用数组来作为表示环形队列的容器int cap;                            // 环形队列的基本容量sem_t data_sem;                     // 给消费者使用的数据资源信号量sem_t space_sem;                    // 给生产者使用的空间资源信号量int consumer_step;                  // 记录消费者要往环形队列中 拿 数据的位置int producer_step;                  // 记录生产者要从环形队列中 放 数据的位置public:ring_queue(int _cap = default_cap) : q(_cap), cap(_cap){sem_init(&data_sem, 0, 0);      // 初始化数据信号量，数据信号量的初始值为 0sem_init(&space_sem, 0, cap);   // 初始化空间信号量，空间信号量的初始值为环形队列的大小consumer_step = 0;              // 消费者线程最开始从环形队列中 拿 数据的位置producer_step = 0;              // 生产者线程最开始往环形队列中 放 数据的位置}// 生产者线程往环形队列中添加数据void put_data(const T &data){sem_wait(&space_sem);           // P 操作: 少了一个空间资源，空间信号量的值 - 1pthread_mutex_lock(&p_mutex);   // 上锁, 所有申请到信号量的线程中，只允许一个去添加数据q[consumer_step] = data;        // 往环形队列中添加数据consumer_step++;                // 走到下一个要放数据的位置consumer_step %= cap;           // 防止越界pthread_mutex_unlock(&p_mutex); // 解锁sem_post(&data_sem);            // V 操作: 多个一个数据资源，数据信号量的值 + 1}// 消费者线程从环形队列中获取数据void get_data(T* data){sem_wait(&data_sem);            // P 操作: 少了一个数据资源，数据信号量的值 - 1pthread_mutex_lock(&c_mutex);   // 上锁, 所有申请到信号量的线程中，只允许一个去获取数据*data = q[producer_step];       // 从环形队列中拿取数据producer_step++;                // 走到下一个要拿数据的位置producer_step %= cap;           // 防止越界pthread_mutex_unlock(&c_mutex); // 解锁sem_post(&space_sem);           // V 操作: 多了一个空间资源，空间信号量的值 + 1}~ring_queue(){sem_destroy(&data_sem);         // 销毁消费者使用的数据信号量sem_destroy(&space_sem);        // 销毁生产者使用的空间信号量}
};// 消费者线程调用的函数
void *consumer(void *arg)
{int data;ring_queue<int> *rqp = (ring_queue<int> *)arg;while (true){rqp->get_data(&data);   // 从环形队列中获取数据// 显示器也是临界资源，也需要用锁保护起来pthread_mutex_lock(&p_mutex);cout << "consumer get data: " << data << endl;pthread_mutex_unlock(&p_mutex);sleep(1);}
}// 生产者线程调用的函数
void *producer(void *arg)
{ring_queue<int> *rqp = (ring_queue<int> *)arg;srand((unsigned long)time(nullptr));while (true){int data = rand() % 1024;rqp->put_data(data);    // 往环形队列中添加数据// 显示器也是临界资源，也需要用锁保护起来pthread_mutex_lock(&c_mutex);cout << "producer put data: " << data << endl;pthread_mutex_unlock(&c_mutex);sleep(1);}
}// 主线程
int main()
{ring_queue<int> rq; // 定义环形队列对象pthread_t cid;      // 记录消费者线程的 IDpthread_t pid;      // 记录生产者线程的 ID// 创建生产者和消费者线程，并让两种线程调用指定函数pthread_create(&cid, nullptr, consumer, (void *)&rq);   // 创建消费者线程pthread_create(&pid, nullptr, producer, (void *)&rq);   // 创建生产者线程pthread_join(cid, nullptr); // 主线程等待消费者线程退出pthread_join(pid, nullptr); // 主线程等待生产者线程退出return 0;
}