基于信号量的生产者消费者模型

news2024/12/28 21:00:42

文章目录

信号量
- 认识概念
- 基于线程分析信号量
- 信号量操作
循环队列下的生产者消费者模型
- 理论认识
- 代码部分

信号量

认识概念

信号量本质:
计数器
它也叫做公共资源

为了线程之间,进程间通信------>多个执行流看到的同一份资源---->多个资源都会并发访问这个资源(此时易出现覆盖)---->数据不一致的问题----->保护起来---->互斥和同步

总结:解决问题的同时会伴随新问题的发生
互斥:任何时候只允许一个执行流(进程)访问公共资源,加锁完成
同步:多个执行流执行的时候,按照一定的顺序执行

被保护起来的资源,临界资源(如,管道)<------->非临界资源
访问该临界资源的代码叫做临界区<----->非临界区

维护临界资源就是维护临界区

原子性：只有两态,对于一件事要么没做要么就做完了(操作系统中)

如何理解信号量?
看电影为例,买票座位属于你,还是坐上去座位就属于你?
但有人占座,这个时候电影院内部的座位,就是多人共享的资源------公共资源
买票的本质: 是对资源的预订机制

有100个座位就绝对不会卖出101张票,如何做到这一点?
	维护一个计数器,
	int count = 100;//表示公共资源的个数

在这里插入图片描述
信号量:表示对资源数目的计数器,每一个执行流系想访问公共资源内的谋一份资源,不应该让执行流直接访问
而是先申请信号量资源.先对信号量计数器进行–操作,只要–成功了,就完成了对资源的预定机制
如果申请不成功,执行流被挂起阻塞

对于公共资源,在他的前面添加
在这里插入图片描述对于信号量为1的公共资源
int sem=1;
二元信号量—互斥锁—完成互斥功能
只能有一个人成功 ,其他执行流(进程)访问会出现阻塞

这样的表明允许访问公共资源的计数叫做信号量(信号灯)

分析细节问题:

1.每个进程都要看到同一个信号量资源-----就只能由OS提供,在IPC体系

2.信号量本质也是公共资源(还好信号量的访问并不复杂,只有-- ++ 的操作,如果访问出错,只要阻塞就可以)
这个操作也属于原子性
- - P
++ V
为什么不定义一个int来完成这个操作?
访问数据的不同进程间的操作,包含大量的拷贝等工作,所以单靠一个int不行

3.单个信号量(目前就这么理解)
struct sem
{
int count;//计数器
task_struct *wait_queue;//等待队列,对这个队列里面的进程进行操作(类似需要操作就对他进行)
}

基于线程分析信号量

现在基于线程概念再来整体理解一下:

1.信号量的本质是一把计数器
2.申请信号的本质就是预订资源
3.pv操作是原子的

怎么理解?

在上篇写的阻塞队列是一个公共资源,同时他的访问是一个整体形式去访问
现在假设这个公共资源是一个数组,那么不同线程可以访问不同数组的不同索引由此达到共同访问临界资源的目的.假设这个数组就是大小为7,有8个线程访问这个临界资源就会出现问题,所以这个时信号量就会起作用,他就相当于是一个计数器

所以上述情况,线程在访问之前都会先申请一个信号量,然后访问指定的一个位置(程序员进行维护),访问结束,释放信号量

这时,对于资源的访问判断已经由信号量充当了,所以不需要进行资源就绪的判断信号量申请成功,这个资源就一定能访问,这个也是原子性的
信号量为1,表示这个资源整体就是只能一个线程进行访问,这边就是互斥的

信号量操作

信号量的操作:
1.快速认识接口
信号量创建:

在这里插入图片描述

参数1:定义一个类似pthread_t类型的变量,这个类型在这边是sem_t类型,
参数2:在一个进程的线程之间共享(0表示这个种情况)还是在进程之间共享
参数3:这个信号量计数器的初始值
信号量销毁:

在这里插入图片描述
销毁信号量
信号量申请,也可以说是信号量等待(阻塞等待,有信号量才会进行后续操作,信号量的值减1)

在这里插入图片描述
这边只讨论普通阻塞方式
发布信号量:

在这里插入图片描述
表示信号量资源使用完毕,可以归还资源,将信号量的值+1

循环队列下的生产者消费者模型

理论认识

基于环形队列的生产者消费者问题–理论
环形队列:
逻辑上为环状,入队列出队列为同一个位置
判空判满可以加一个计数器或者是消耗一个空间的方式进行
这边不做讨论,因为信号量的操作就能完成判定
在这里插入图片描述
在循环队列中,

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

消费者生产者在这个结构中也是一样,为空或者为满会指向同一个位置,这个时候就不能并发访问,在其他情况下可以去并发访问,也就是并发进入临界区.

所以在为空为满为互斥,生产者消费者跑是同步,这是需要局部维持的"资源"的认识:
P:空间是资源(无空间了,不生产)
c:数据是资源(无数据了,不消费)
所以需要两个信号量来维护这个资源

伪代码:

p->sem_space:N
c->sem_data:0
生产者:
	p(sem_space)//预定资源,申请空间资源用来生产
	//生产行为,位置占据
	v(sem_data)//告诉消费者可以进行消费		
消费者:
	p(sem_data)//申请数据资源用来消费
	//消费行为
	v(sem_space)//告诉生产者当前位置为空可以来生产

代码部分

单任务的循环队列式生产者消费者模型
main.cc

#include "RingQueue.hpp"
#include <unistd.h>
void *Productor(void *args)
{
    RingQueue<int> *rq = static_cast<RingQueue<int> *>(args);
    int cnt = 100;
    while(true)
    {
        rq->Push(cnt);
        std::cout << "Productor done, the data is " << cnt << std::endl;
        cnt--;
    }
}
void *Consumer(void *args)
{
    RingQueue<int> *rq = static_cast<RingQueue<int> *>(args);
    while(true)
    {
        int data = 0;
        rq->Pop(&data);
        std::cout << "Consumer done, the data is " << data << std::endl;
        sleep(1);
    }
    
}
int main()
{
    pthread_t c, p;
    RingQueue<int> *rq = new RingQueue<int>();
    pthread_create(&p, nullptr, Productor, rq);
    pthread_create(&c, nullptr, Consumer, rq);   

    pthread_join(p, nullptr);
    pthread_join(c, nullptr);


    return 0;
}

RingQueue.hpp

#pragma once
#include <iostream>
#include <vector>
#include <semaphore.h>
#include <pthread.h>

const int defaultsize = 5;
template<class T>
class RingQueue
{
private:
    void P(sem_t &sem)//用于申请可以访问的资源
    {
        sem_wait(&sem);
    }
    void V(sem_t &sem)//资源使用完毕
    {
        sem_post(&sem);
    }
public:
    RingQueue(int size = defaultsize):_ringqueue(size), _size(size), _p_step(0), _c_step(0)
    {
        sem_init(&_space_sem, 0, size);
        sem_init(&_data_sem, 0, 0);
    }
    void Push(const T& in)//生产者放入数据
    {
        P(_space_sem);
        _ringqueue[_p_step] = in;
        _p_step++;
        _p_step %= _size;
        V(_data_sem);
    }
    void Pop(T *out)
    {
        P(_data_sem);
        *out = _ringqueue[_c_step];
        _c_step++;
        _c_step %= _size;
        V(_space_sem);
    }
    ~RingQueue()
    {
        sem_destroy(&_space_sem);
        sem_destroy(&_data_sem);
    }
private:
    std::vector<T> _ringqueue;
    int _size;

    sem_t _space_sem;
    sem_t _data_sem; 

    int _p_step;
    int _c_step;
};