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

[Linux]生产消费者模型

news 来源：原创 2025/5/6 11:02:17

一、生产消费者模型概念

1.概念

2.模块构成

3.协作机制

二、基于BlockingQueue的多CP问题

1.BlockQueue模块

2.Task任务模块

3.创建多线程以及整合模块

三、基于信号量的环形队列CP模型

1.POISX信号量接口

初始化信号量

PV操作

信号量销毁

2.模型简述

3.代码实现

一、生产消费者模型概念

1.概念

生产者消费者模型按名字来听就一定有生产者和消费者成员，生产者将自己生产的物品交给消费者，这就是最简单的生产消费协同。但是这样的话，生产者生产之后就一定要当面交给消费者，然后消费者不接收生产者就无法生产，这样就会有弊端了。如果说破生产者生产的快，但是消费者消费速度慢，那么生产者就需要等待消费者将东西买走后，才能继续生产，就会大大降低产能和效率。

所以接下来引入了超市的这一中间概念，生产者可以将生产的东西放到超市里面，不需要直接给消费者，这样的话，生产者的产能就不会受到消费者的影响了，而且消费者也可能随时去超时获取自己想要的东西，而不是说生产者生产出来，消费者就必须接收了。这个就是生产消费者模型的大概框架。

生产消费者模式是一种经典的多线程或多进程并发协作的一种模式。他主要用于解决数据的产生和消费速度不一致的问题，实现了功能解耦、时间解耦以及处理速度的解耦，同时保证了数据的完整性和正确性。功能解耦：生产者专注生产，不关心数据如何被消费者处理；消费者专注处理，不关心数据如何生产的。时间解耦：生产者和消费者(CP)之间不需要同步运行，在满足各自条件下进行工作就可以。处理速度解耦：CP之间不需要互相等待对方，所以处理速度快的一方那么就会一直以高速的方式工作，而慢速的一方则慢慢执行即可，有中间的共享资源区作为数据的缓冲地带。

2.模块构成

分为生产者、消费者以及缓冲区三部分，生产者和消费者可以是一个线程、进程或者软件模块，生产者将生产的数据放入缓冲区，消费者从缓冲区中取出数据后做相应的处理工作。

3.协作机制

该模型一定涉及到执行流的互斥与同步问题了，生产者与消费者之间存在互斥关系，生产者访问缓冲区的时候，消费者是不可以访问的，消费者访问的时候，同样的生产者也是不可以访问的。因为生产和消费二者访问缓冲区的时候，会对缓冲区的数据做修改，那么就会有线程安全问题，所以要实现互斥机制。当然二者之间也有同步关系，缓冲区中没有数据时，消费者不能进行访问缓冲区，当生产者生产数据之后，会通知消费者访问的。

对于多生产多消费的情况下，也就是生产者和消费者都有很多线程。生产和消费都会修改缓冲区。所以生产者与生产者之间，消费者与消费者之间，也属于多线程访问共享资源，也是要实现互斥的。

上述概括下来，就是生产者之间、消费者之间的互斥以及生产者消费者之间的互斥与同步。

二、基于BlockingQueue的多CP问题

1.BlockQueue模块

该模块定义了一个任务队列，也就是CP问题的缓冲区部分、以及生产线程和消费线程对于该缓冲区的操作方法。当生产者要放入任务的时候，要先判断缓冲区中是否有空间，如果有空间的话，就放入任务，并且通知消费进程取出任务，当没有空间的时候，就进行等待条件变量就绪，这个条件变量的唤醒是由消费进程取出数据后，唤醒的，唤醒生产进程放入数据。

对于消费进程要取出任务的时候，首先判断缓冲区有没有任务，如果有的话，就取出任务，并通知生产进程放入数据，也就是唤醒当时因为缓冲区满了，而阻塞在条件变量中的等待线程。当没有任务的时候，就阻塞在条件变量中等待生产进程唤醒。

生产线程和消费线程之间通过锁实现的互斥操作，通过条件变量实现了同步的顺序操作。

#pragma once#include <iostream>
#include <queue>
#include <pthread.h>// 缓冲区默认容量
static const int default_capacity = 5;//生产消费者模型类
template <class T>
class BlockQueue
{
private:std::queue<T> _task_queue; // 任务队列int _capacity;             // 队列容量pthread_mutex_t _mutex;    // 锁pthread_cond_t _c_cond;    // customer条件变量pthread_cond_t _p_cond;    // productor条件变量public:// 构造函数BlockQueue(int capacity = default_capacity):_capacity(capacity){//初始化锁和条件变量pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_c_cond, nullptr);pthread_cond_init(&_p_cond, nullptr);}//生产者--将任务放入任务队列void Push(const T& task){//加锁pthread_mutex_lock(&_mutex);//判断缓冲区是否满了,如果满了的话，就一直通知消费者取出数据//使用while是防止伪唤醒。while(_capacity == _task_queue.size()){//释放锁，并等待消费者通知可以生产pthread_cond_wait(&_p_cond, &_mutex);}//有空间了，放入任务_task_queue.push(task);//通知消费者可以取数据了pthread_cond_signal(&_c_cond);//解锁pthread_mutex_unlock(&_mutex);}//消费者--将任务从任务队列取出void PoP(T* task){//加锁pthread_mutex_lock(&_mutex);//判断是否有数据，没有就进行等待while(_task_queue.empty()){//释放锁，并等待生产者通知可以消费pthread_cond_wait(&_c_cond, &_mutex);}//有数据了，取出数据*task = _task_queue.front();_task_queue.pop();//通知生产者可以生产了pthread_cond_signal(&_p_cond);//解锁pthread_mutex_unlock(&_mutex);}//析构函数~BlockQueue(){//锁和条件变量的销毁pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_c_cond);pthread_cond_destroy(&_p_cond);}
};

2.Task任务模块

定义的是计算两个元素的加减法类

#pragma once#include <iostream>
#include <unistd.h>
#include <string>static const int defaultValue = 0;
static const std::string str = "+-";// 计算加减法的任务类
class Task
{
private:int _x;int _y;char _op;int _result;public://构造与析构函数Task() {}Task(int x, int y, char op): _x(x), _y(y), _op(op), _result(defaultValue) {}~Task() {}void Run(){switch (_op){case '+':_result = _x + _y;break;case '-':_result = _x - _y;break;default:std::cout << "op is error" << std::endl;break;}}// 运行任务重载函数void operator()(){Run();std::cout << _x << _op << _y << "=" << _result << std::endl;sleep(2);}
};

3.创建多线程以及整合模块

该模块则是定义了生产线程和消费线程具体线程函数。以及创建线程、定义任务队列类等操作。

#include <cstdlib>
#include <vector>
#include <string>
#include <cstring>
#include <pthread.h>
#include "BlockQueue.hpp"
#include "Task.hpp"// 线程传递的参数
struct ThreadData
{BlockQueue<Task> *_cp;    // 生产者消费者模型对象std::string _thread_name; // 线程名称// 构造函数ThreadData(BlockQueue<Task> *cp, std::string name): _cp(cp), _thread_name(name) {}
};
// 消费者线程函数
void *Costomer(void *arg)
{// 接收参数ThreadData *data = (ThreadData *)arg;while (true){// 获取任务Task task;data->_cp->PoP(&task);std::cout << data->_thread_name << " run task : ";// 执行任务task();}return nullptr;
}
// 生产者线程参数
void *Productor(void *arg)
{// 接收参数ThreadData *data = (ThreadData *)arg;while (true){// 生产任务数据int data1 = rand() % 10;int data2 = rand() % 10;char op = str[rand() % str.size()];Task task(data1, data2, op);// 将任务放入队列data->_cp->Push(task);std::cout << data->_thread_name << " push task into queue : " << data1 << op << data2 << std::endl;}return nullptr;
}int main()
{// 生产随机数种子srand((uint16_t)time(nullptr) ^ getpid() ^ pthread_self());// 创建生产者消费者模型对象BlockQueue<Task> *cp = new BlockQueue<Task>();// 创建生产者消费者线程std::vector<pthread_t> custumer_thread(5);std::vector<pthread_t> productor_thread(5);// 消费者for (int i = 0; i < 5; i++){// 线程idpthread_t tid;// 构造线程参数std::string thread_name = "customer-" + std::to_string(i);ThreadData data(cp, thread_name);// 创建线程pthread_create(&tid, nullptr, Costomer, &data);// 放入数组中custumer_thread.push_back(tid);}// 生产者for (int i = 0; i < 5; i++){// 线程idpthread_t tid;// 构造线程参数std::string thread_name = "productor-" + std::to_string(i);ThreadData data(cp, thread_name);// 创建线程pthread_create(&tid, nullptr, Productor, &data);// 放入数组中productor_thread.push_back(tid);}// 线程等待for (size_t i = 0; i < custumer_thread.size(); ++i){if (pthread_join(custumer_thread[i], nullptr) != 0){std::cerr << "等待消费者线程结束失败" << std::endl;}}for (size_t i = 0; i < productor_thread.size(); ++i){if (pthread_join(productor_thread[i], nullptr) != 0){std::cerr << "等待生产者线程结束失败" << std::endl;}}// 释放资源delete cp;return 0;
}

三、基于信号量的环形队列CP模型

1.POISX信号量接口

其实和System V信号量的作用基本上是一样的，只是接口上有区别。

初始化信号量

int sem_init(sem_t* sem, int pshared, unsigned int value);

第一个参数是传递要初始化的信号量；第二个参数表示信号量是用于进程间（非零值）还是线程间（零值）共享；第三个参数则是信号量的初始值。

PV操作

int sem_wait(sem_t* sem); //申请操作

int sem_post(sem_t* sem); //释放操作

信号量销毁

int sem_destory(sem_T* sem);

2.模型简述

该模型使用了锁实现了线程之间的互斥，信号量的方式实现了同步机制。但是这里的互斥只有生产者和生产者、消费者和消费者之间的互斥，没有生产者与消费者之间的互斥。因为使用了信号量记录了该资源区域的使用情况，用信号量表示了剩余空间的个数和存在的资源的个数，相当于将一个大的资源区域分成了若干个小区域。同时配合指针记录生产者下一次放数据的位置，消费者下一次取数据的位置。所以生产者线程和消费者线程两者访问不到同一个位置，所以也就不存在两个的互斥问题了。

信号量实现的同步是，当space剩余空间信号量为0的时候，生产线程无法放入数据，会等待消费线程取出数据，当消费线程取出数据后，会释放一个space信号量，那么生产线程就可以申请到信号量了，就可以放入数据了。放入数据之后，会将data信号量释放一个，表示着队列中的数据量多了一个。

在申请信号量和锁的顺序上，应该是先申请信号量，信号量代表的是该线程对于想要获取的资源是否存在，如果存在的话，再去为了访问资源而申请锁，如果说没有资源的话，申请到锁也没用，所以说应该先去申请信号量，之后再去申请锁资源。而且先申请锁的话，只会有一个线程到达申请信号量的步骤，而先申请信号量的话，可以有多个线程到达申请锁的位置等待锁资源。

经过上述阐述，因为生产者和消费者没有互斥的关系，所以要给生产者和消费者各设定一把锁去保证线程安全。同时还有设定两个信号量，一个是剩余空间信号量，另一个是剩余资源的信号量。

3.代码实现

#pragma once#include <iostream>
#include <pthread.h>
#include <vector>
#include <semaphore.h>// 环形队列默认大小
static const int default_capacity = 5;template <class T>
class RingQueue
{
private:std::vector<T> _ringQueue; // 环形队列int _capacity;             // 最大容量int _p_step;               // 生产者放入任务的位置int _c_step;               // 消费者取出任务的位置sem_t _space_sem;          // 剩余空间信号量sem_t _data_sem;           // 剩余任务信号量pthread_mutex_t _p_mutex;  // 生产锁pthread_mutex_t _c_mutex;  // 消费锁private://申请信号量操作void P(sem_t &sem) { sem_wait(&sem); }//释放信号量操作void V(sem_t &sem) { sem_post(&sem); }public://构造函数RingQueue(int capacity = default_capacity):_ringQueue(capacity), _capacity(capacity), _p_step(0), _c_step(0){//初始化信号量，剩余空间为capacity，剩余任务数据为0sem_init(&_space_sem, 0, _capacity);sem_init(&_data_sem, 0, 0);//初始化锁pthread_mutex_init(&_p_mutex, nullptr);pthread_mutex_init(&_c_mutex, nullptr);}//生产者放入任务操作void Push(const T& task){//申请信号量并加锁P(_space_sem);pthread_mutex_lock(&_p_mutex);//放入任务_ringQueue[_p_step] = task;//更新生产者指针位置_p_step++;_p_step %= _capacity;//解锁并释放信号量pthread_mutex_unlock(&_p_mutex);V(_data_sem);}//消费者取出任务操作void Pop(T* task){//申请信号量并加锁P(_data_sem);pthread_mutex_lock(&_c_mutex);//取出任务*task = _ringQueue[_c_step];//更新消费者指针位置_c_step++;_c_step %= _capacity;//解锁并释放信号量pthread_mutex_unlock(&_c_mutex);V(_space_sem);}//析构函数~RingQueue(){//销毁信号量和锁sem_destroy(&_space_sem);sem_destroy(&_data_sem);pthread_mutex_destroy(&_p_mutex);pthread_mutex_destroy(&_c_mutex);}
};

一、生产消费者模型概念

1.概念

2.模块构成

3.协作机制

二、基于BlockingQueue的多CP问题

1.BlockQueue模块

2.Task任务模块

3.创建多线程以及整合模块

三、基于信号量的环形队列CP模型

1.POISX信号量接口

初始化信号量

PV操作

信号量销毁

2.模型简述

3.代码实现

相关文章：