当前位置：首页 > news >正文

[Linux——Lesson25.线程：生产与消费者模型]

news 2025/11/15 7:33:33

前言

一、😗认识生产消费者模型

1-1🍕为什么使用生产消费者模型

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

2-1 🍿单生产单消费示例

2-1-1 🦣Thread.hpp线程类封装

2-1-2 🐕BlockQueue.hpp阻塞队列类封装

2-1-3🌟main函数逻辑

2-1-4 🐈‍⬛运行结果及拓展

2-2 🥐生产消费者模型的理解

三、😋循环队列生产消费模型

3-1 🌭POSIX 信号量

3-2 🥨二元信号量

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

🗝️总结与提炼

1️⃣核心本质与价值

2️⃣两种实现方案对比

3️⃣关键技术点梳理

4️⃣实践启示与注意事项

5️⃣核心收获

结束语

前言

在多线程并发编程的领域中，线程间的高效协作与资源协调是构建稳定、高效系统的核心难题。当多个线程共享资源并协同完成任务时，如何避免访问冲突、解决线程执行速度差异引发的冲突，以及实现安全的信息交互，成为开发者必须攻克的关键问题。

生产消费者模型是我们解决多线程同步互斥问题的经典模型，为上述挑战提供了合适的解决方案。该模型通过引入 “生产者”“消费者” 两种角色与共享缓冲区，将线程任务拆解为数据生成、存储与消费的闭环流程：生产者线程负责生成数据并放入缓冲区，消费者线程从缓冲区取出数据进行处理，而同步互斥机制则确保了缓冲区操作的原子性与顺序性，有效避免了数据不一致，临界资源访问冲突等问题。

深入剖析生产消费者模型的设计思想对我们进行多线程编程是百利而无一害的。

本文将从3方面解析生产消费者模型：

计算机中的生产消费者模型；
基于阻塞队列的生产消费者模型；
基于环形队列的生产消费者模型。

一、😗认识生产消费者模型

生产消费者模型的本质是讨论数据如何并发的传递的问题，生活当中其实有很多生产消模型，经典模型比如超市。超市本身是不生产产品的，商品源来自于厂商，而客户源就是我们普通老百姓。而厂商可能不止一家，用户也不止一个，那为什么说超市是经典的模型呢？

以线程的角度来说，超市其实就是 共享资源，而厂商和用户其实就是多个线程，那么这个超市就要考虑多线程的同步和互斥问题，如果临界资源是超市，那么超市的商品毫无疑问就是数据，既然商品是数据，那么存储着数据的超市，不就是临时保存数据的 “内存空间” 吗？

实际上确实如此，而这里的 “超市” 并不是指硬件上的物理内存，其 是一种数据结构类型，可以是队列，可以是栈结构等等。总的来说，“超市” 是 “数据（商品）” 的交易场所。

既然超市一边有供货商提供货源，一边有用户来消费购物，OS层面上来说，这个临街资源两边都有多个线程，而我们 把 “供货商” 线程称为生产者，“客户” 称为消费者。

那么毫无疑问生产消费者模型是一个多线程访问共享资源的场景，在之前我们曾学习过，如果多线程不加保护的同时访问临界资源会带来严重的后果，一个简单的例子就是带来数据不一致的问题。要了解问题，必究其原因，所以我们需要先把它们的关系理清：

①生产者与生产者之间：互斥。
②生产者与消费者之间：互斥。
③消费者与消费者之间：互斥、同步。

三者皆是互斥关系，其实很好理解，假如超市只剩下了一件商品，而需要这件商品的客户却不止一个，那么这些客户之间就是竞争关系，在OS层面来说叫做互斥关系，同理，超市的货架是有限的，那么生产商会想办法把自己的商品更多的放在货架上让消费者看到，所以他们也是互斥。而当供应商需要到货100件之后才能对超市的账，但是在100件之前如果被客户直接拿走了，很可能会导致后面对账对不上等问题，所以生产者与消费者之间其实也是互斥关系的。如果有一批客户非常需要一件商品，但是这件商品迟迟不上架，所以超市就让他们两个保持距离，当供应商供货的时候让客户来购买。保持的这种关系我们也称之为同步。
注：生产者与生产者，消费者于消费者之间并不绝对的保持互斥，如果你想人为的保持同步也是可以的。

综上所述，生产消费者之间存在3种关系(上面三个关系)，2个角色（生产者角色和消费者角色）以及一个交易场所（超市）。你可以称之为 “321”原则（仅用来方便记忆）。

1-1🍕为什么使用生产消费者模型

生产消费者模型主要解决了：生产者与消费者的强耦合关系，生产者在生产数据的时候不需要等消费者使用，而是直接将数据放到缓冲区中；同样消费者要使用数据也不会直接找生产者要，而是直接从缓冲区中进行读取。通过中间的缓冲区将生产者和消费者进行解耦。

通过中间的缓冲区，生产消费者模型有了诸多优势：

解耦：生产者和消费者无需知道对方的存在，只需通过缓冲区交互，降低模块间耦合度
削峰填谷：当生产速度超过消费速度时，缓冲区可以暂存数据；当消费速度快时，也能从缓冲区获取数据
并发协作：支持并发，此处的并发指的是：允许多个生产者一起生产数据，多个消费者一起消费数据；但是不论是向缓冲区写，还是向缓冲区中读，都要保证是线程安全的。
负载均衡：多个生产者可以向同一缓冲区提交任务，多个消费者可以从缓冲区获取任务处理

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

上面我们说了，生产消费者模型中的临界资源是一种数据结构，而比较常见的一种是基于阻塞队列的生产消费者模型：

将队列作为中间的缓冲区，生产者写入的数据放到队列中，消费者从队列中拿数据。

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

2-1 🍿单生产单消费示例

2-1-1 🦣Thread.hpp线程类封装

我们要实现生产消费者模型，自己对线程函数进行封装，达到方面的目的：

#ifndef __THREAD_HPP__
#define __THREAD_HPP__#include <iostream>
#include <string>
#include <unistd.h>
#include <functional>
#include <pthread.h>namespace ThreadModule
{template<typename T>using func_t = std::function<void(T&)>;// 重命名，把将来线程需要执行的回调函数进行包装// typedef std::function<void(const T&)> func_t;template<typename T>class Thread{public:void Excute(){_func(_data);}public:Thread(func_t<T> func, T &data, const std::string &name="none-name"): _func(func), _data(data), _threadname(name), _stop(true){}static void* threadroutine(void *args) // 类成员函数，形参是有this指针的！！而静态成员函数没有this指针{Thread<T> *self = static_cast<Thread<T> *>(args);// 将args强转为Thread<T>*self->Excute();// 执行回调return nullptr;}bool Start(){int n = pthread_create(&_tid, nullptr, threadroutine, this);if(!n){_stop = false;return true;}else{return false;}}void Detach(){if(!_stop){pthread_detach(_tid);}}void Join(){if(!_stop){pthread_join(_tid, nullptr);}}std::string name(){return _threadname;}void Stop(){_stop = true;}~Thread() {}private:pthread_t _tid;std::string _threadname;T &_data;  // 为了让所有的线程访问同一个全局变量func_t<T> _func;bool _stop;};
} // namespace ThreadModule#endif

以上的Thread.hpp的类，包含了线程常用用途，包括：线程创建，线程等待，线程分离。以及线程所需要的共享资源，threadroutine()为创建线程时的入口函数（回调），通过该回调函数，可以调用使用function包装的自定义函数，这样也就实现了让不同线程执行不同的自定义任务功能。

2-1-2 🐕BlockQueue.hpp阻塞队列类封装

有了线程类，我们还需要一个阻塞队列用来作为生产消费模型的场所，实现如下阻塞队列：
#ifndef __BLOCK_QUEUE_HPP__
#define __BLOCK_QUEUE_HPP__#include<iostream>
#include <string>
#include <pthread.h>
#include <queue>template<typename T>
class BlockQueue
{
private:bool IsFull()// 判满{return _block_queue.size() == _cap;}bool IsEmpty()// 判空{return _block_queue.empty();}
public:BlockQueue(int cap): _cap(cap){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_product_cond, nullptr);pthread_cond_init(&_consum_cond, nullptr);}void Enqueue(T& in)// 生产者{pthread_mutex_lock(&_mutex);// if(IsFull())// 判满// {//     pthread_cond_wait(&_product_cond, &_mutex);// }while(IsFull())// 判满{// 之前 ： 安全_product_wait_num++;pthread_cond_wait(&_product_cond, &_mutex);_product_wait_num--;// 之后 ： 安全}// 生产_block_queue.push(in);// 通知消费者消费if(_consumer_wait_num > 0)pthread_cond_signal(&_consum_cond);// pthread_cond_signal(): 全部唤醒pthread_mutex_unlock(&_mutex);}void Pop(T *out)// 消费者{pthread_mutex_lock(&_mutex);// if(IsEmpty())// {//     pthread_cond_wait(&_consum_cond, &_mutex);// }while(IsEmpty()){_consumer_wait_num++;pthread_cond_wait(&_consum_cond, &_mutex);_consumer_wait_num--;}// 进行消费*out = _block_queue.front();_block_queue.pop();// 通知生产者生产if(_product_wait_num > 0)pthread_cond_signal(&_product_cond);pthread_mutex_unlock(&_mutex);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_product_cond);pthread_cond_destroy(&_consum_cond);}
private:std::queue<T> _block_queue;pthread_mutex_t _mutex;// 保护blockqueue的锁pthread_cond_t _product_cond;// 专门给生产者提供的条件变量pthread_cond_t _consum_cond;// 专门给消费者提供的条件变量int _cap;// 容量int _consumer_wait_num;// 消费者等待数量int _product_wait_num;// 生产者等待数量
};#endif
一个基于阻塞队列的生产消费者模型，使用线程模拟，而阻塞队列作为临街资源，所以线程免不了要进行加锁解锁等待等操作。所以在private成员内，定义了生产消费者的条件变量与互斥锁。

构造函数来初始化锁和互斥量，以及初始化队列容量。队列的操作我们不陌生，无非就是出队列和入队列。这里出队列相当于消费者正在消费，入队列相当于生产者在生产。所以pop和enqueue操作就是临界区。所以在访问接口之前需要加锁。

如果队列满了，生产者需要判满，将该线程进行条件变量等待。只有当消费者消费过后，队列不再满，才能从消费者处进行唤醒。同理，如果队列为空，消费者不可进行消费，进行判空，当生产者生产出数据时，再唤醒消费者进行消费。

这里还有一个细节，我们知道当线程进行条件变量唤入等待队列之后，如果该线程被唤醒，会重新竞争锁，竞争成功后，从上次阻塞的地方继续运行下去。但是这里有一个坑，如果此时生产消费模型是单生产多消费模型，线程唤醒方式为broadcast(全唤醒)，假设此时生产者只来得及生产一个数据，但是此时所有的消费者都被唤醒。只有一个线程竞争到了锁，那么这个线程就可消费这唯一的资源，当消费完成后就会归还锁。
这没有问题，但是我还有其他线程在竞争锁啊！！此时第一个线程归还了锁之后，其他线程又抢到了锁，但是这个时候阻塞队列里确为空，空队列里pop()就会报错。这种行为我们称之为伪唤醒。为了防止伪唤醒，我们在判空和判满的时候不能单纯的使用if判断，直接使用while循环判断就不会导致伪唤醒行为了。

2-1-3🌟main函数逻辑

以上的准备工作做好，我们就可以执手创建一个生产消费模型了：

#include "BlockQueue.hpp"
#include "thread.hpp"
#include <iostream>
#include <string>
#include <vector>
#include <unistd.h>using namespace ThreadModule;
int a = 10;void Consumer(BlockQueue<int> &bq)// 消费者执行自定义函数
{while(true){sleep(5);int data;bq.Pop(&data);std::cout << "Consumer consum data is:  " << data << std::endl;}
}void Productor(BlockQueue<int> &bq)// 生产者自定义函数
{int cnt = 10;while(true){bq.Enqueue(cnt);std::cout << "Productor product data is: " << cnt++ << std::endl;}
}void StartComm(std::vector<Thread<BlockQueue<int>>> *threads, int num, BlockQueue<int> &bq, func_t<BlockQueue<int>> func)
{for(int i = 0; i < num; ++i){std::string name = "Thread-" + std::to_string(i + 1);// 传递线程名称threads->emplace_back(func, bq, name);// 将初始化的线程信息插入进数组threads->back().Start();// 创建线程}
}// 对Consumer函数进行回调
void StartConsumer(std::vector<Thread<BlockQueue<int>>> *threads, int num, BlockQueue<int> &bq)
{StartComm(threads, num, bq, Consumer);
}// 对Productor函数进行回调
void StartProductor(std::vector<Thread<BlockQueue<int>>> *threads, int num, BlockQueue<int> &bq)
{   StartComm(threads, num, bq, Productor);
}// 线程等待
void WaitAllThread(std::vector<Thread<BlockQueue<int>>> &threads)
{for(auto &thread : threads){thread.Join();}
}int main()
{BlockQueue<int> *bq = new BlockQueue<int>(5);// 阻塞队列大小为5std::vector<Thread<BlockQueue<int>>> threads;// 线程集合// 启动线程StartConsumer(&threads, 1, *bq);// 创建消费者线程，1为线程数目StartProductor(&threads, 1, *bq);// 创建生产者线程，1为线程数WaitAllThread(threads);return 0;
}

main函数的逻辑还是非常简单的，使用数组集合所有线程，通过StartComm函数创建线程，而StartConsumer()和StartProductor()函数用来决定创建的是生产者还是消费者。如果是生产者则回调生产者函数，如果是消费者则回调消费者函数。