基于信号量和环形队列的生产者消费者模型

本文主要是介绍基于信号量和环形队列的生产者消费者模型，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

POSIX信号量

POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步。信号量的本质是一个计数器。

信号量本身是一个判断条件，是资源的预定机制，预定在外部，可以不判断资源是否满足，就可以知道内部资源的情况。

信号量接口

初始化信号量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);Link with -pthread.

pshared:0表示线程间共享，非零表示进程间共享
value：信号量初始值

销毁信号量

#include <semaphore.h>int sem_destroy(sem_t *sem);Link with -pthread.

等待信号量

等待信号量，会将信号量的值减1

 int sem_wait(sem_t *sem);

等待成功继续往后执行，资源不足，阻塞在信号量这里

发布信号量

int sem_post(sem_t *sem);

发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。

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

环形队列在物理结构上是一个线性结构，逻辑结构我们可以认为是一个环形结构。

环形队列有一头一为，头部进行取数据，尾部进行放数据，最开始指向同一个位置。

现在进行只插入数据操作，end一直进行++操作，此时end又和head相遇，也就是说，当队列为空或者为满的时候，head和end是相等的。

这样就出现了歧义，head==end无法判断队列的状态，因此引入了计数器或牺牲掉一个空位置（head==end+1表示队列满了）。

上面已经了解了信号量，因此队列空和满不再是本节需要关注的问题，需要关注的是多线程如何在环形队列中进行生产和消费。

当队列为空的时候，理论上只能让生产者先生产；当队列为满的时候，只能让消费者先消费，这就保证在访问的时候有一定的顺序性和互斥特点。
环形队列不为空也不为满时，生产和消费的下标，一定指的不是同一个位置（head!=end），此时允许生产和消费同时进行。

结论：

1. 不能让生产者把消费者套一个圈
2. 不能让消费者超过生产者

通过信号量来完成上述要求，实现同步和互斥。

消费者最关心的是数据资源，生产者最关心的是空间资源。

定义两个信号量：

1. sem_t data_sem = 0数据信号量
2. sem_t space_sem = N空间信号量

作为生产者需要申请空间，执行P操作：P(space_sem) ，生产数据后，空间被占了，执行V操作：V(data_sem)
作为消费者需要申请资源，执行P操作：P(sem_data)，一旦将数据拿走，空间就多出来了，再执行一个V操作：V(spce_sem)
因此生产者和消费者是申请自己的资源，释放对方的资源。

单生产单消费

先将环形队列生产满，然后消费一个，生产一个，体现同步特性：

//RingQueue.hpp
#pragma once
#include<iostream>
#include<string>
#include<vector>
#include<semaphore.h>template<typename T>
class RingQueue
{
private:void P(sem_t &s){sem_wait(&s);}void V(sem_t &s){sem_post(&s);}public:RingQueue(int max_cap):_ringqueue(max_cap),_max_cap(max_cap),_c_step(0),_p_step(0){sem_init(&_data_sem,0,0);sem_init(&_space_sem,0,_max_cap);}//生产者void Push(const T &in){P(_space_sem);_ringqueue[_p_step]=in;_p_step++;_p_step%=_max_cap;V(_data_sem);}//消费者void Pop(T *out){P(_data_sem);*out=_ringqueue[_c_step];_c_step++;_c_step%=_max_cap;V(_space_sem);}~RingQueue(){sem_destroy(&_data_sem);sem_destroy(&_space_sem);}private:std::vector<T> _ringqueue;int _max_cap;int _c_step;int _p_step;sem_t _data_sem;  //消费者信号量sem_t _space_sem;  //生产者消费量
};

//Main.cc
#include"RingQueue.hpp"
#include<iostream>
#include<pthread.h>
#include<ctime>
#include<unistd.h>void *Consumer(void *argc)
{RingQueue<int> *rq=static_cast<RingQueue<int> *>(argc);while(true){sleep(1);int data=0;//1.消费rq->Pop(&data);std::cout<<"Consumer -> "<<data<<std::endl;}
}void *Productor(void *argc)
{RingQueue<int> *rq=static_cast<RingQueue<int> *>(argc);while(true){//1.构造数据int data=rand()%10+1;//2.生产rq->Push(data);std::cout<<"Productor -> "<<data<<std::endl;}
}int main()
{srand(time(nullptr)^getpid());RingQueue<int> *rq=new RingQueue<int>(5);//单生产单消费pthread_t c,p;pthread_create(&c,nullptr,Consumer,rq);pthread_create(&p,nullptr,Productor,rq);pthread_join(c,nullptr);pthread_join(p,nullptr);return 0;
}

运行结果：

生产一个消费一个，体现互斥特点：

//Main.cc
#include"RingQueue.hpp"
#include<iostream>
#include<pthread.h>
#include<ctime>
#include<unistd.h>void *Consumer(void *argc)
{RingQueue<int> *rq=static_cast<RingQueue<int> *>(argc);while(true){int data=0;//1.消费rq->Pop(&data);std::cout<<"Consumer -> "<<data<<std::endl;}
}void *Productor(void *argc)
{RingQueue<int> *rq=static_cast<RingQueue<int> *>(argc);while(true){sleep(1);//1.构造数据int data=rand()%10+1;//2.生产rq->Push(data);std::cout<<"Productor -> "<<data<<std::endl;}
}int main()
{srand(time(nullptr)^getpid());RingQueue<int> *rq=new RingQueue<int>(5);//单生产单消费pthread_t c,p;pthread_create(&c,nullptr,Consumer,rq);pthread_create(&p,nullptr,Productor,rq);pthread_join(c,nullptr);pthread_join(p,nullptr);return 0;
}

上述测试代码是传递一个int类型的数据到阻塞队列中，也可以传递其他类型，在传递struct或者class类型时，可以封装成一个个的任务传递到环形中。

传递一个任务：

//Task.hpp
#pragma once
#include<iostream>
#include<string>
#include<functional>class Task
{public:Task(){}Task(int x,int y):_x(x),_y(y){}void Excute(){_result=_x+_y;}void operator()(){Excute();}std::string debug(){std::string msg=std::to_string(_x)+"+"+std::to_string(_y)+"=?";return msg;}std::string result(){std::string msg=std::to_string(_x)+"+"+std::to_string(_y)+"="+std::to_string(_result);return msg;}private:int _x;int _y;int _result;
};

//Main.cc
#include"RingQueue.hpp"
#include"Task.hpp"
#include<iostream>
#include<pthread.h>
#include<ctime>
#include<unistd.h>
//传递任务
void *Consumer(void *argc)
{RingQueue<Task> *rq=static_cast<RingQueue<Task> *>(argc);while(true){Task t;//1.消费rq->Pop(&t);t();std::cout<<"Consumer -> "<<t.result()<<std::endl;}
}void *Productor(void *argc)
{RingQueue<Task> *rq=static_cast<RingQueue<Task> *>(argc);while(true){sleep(1);//1.构造数据int x=rand()%10+1;usleep(x*1000);int y=rand()%10+1;Task t(x,y);//2.生产rq->Push(t);std::cout<<"Productor -> "<<t.debug()<<std::endl;}
}int main()
{srand(time(nullptr)^getpid());RingQueue<Task> *rq=new RingQueue<Task>(5);//单生产单消费pthread_t c,p;pthread_create(&c,nullptr,Consumer,rq);pthread_create(&p,nullptr,Productor,rq);pthread_join(c,nullptr);pthread_join(p,nullptr);return 0;
}

运行结果：

多生产多消费

生产者和消费者的下标位置只有一个，在环形队列中，执行多生产多消费操作，一瞬间下标称为自己的临界资源，所以必须要加锁。多生产多消费是为了在处理放数据取数据有更好的并发度。

先申请锁合适还是先申请信号量合适？
如果先加锁，申请信号量的线程就是一个生产者，一旦解锁，线程又得重新申请信号量，效率地下，申请锁和申请信号量注定是串行的。如果是先申请信号量，先预定着，然后再去竞争，谁的优先级高谁就先申请到锁。这就类似于我们日常生活中现在手机上面购票，等电影快开始准备检票即可。**因此先申请信号量在加锁合适。**申请信号量本身是原子的，不会出错，先把可用的资源给线程瓜分，然后等待即可。

---

这篇关于基于信号量和环形队列的生产者消费者模型的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---