Linux —— 信号量
- 什么是信号量
- P操作(Wait操作)
- V操作(Signal操作)
- 信号量的类型
- 一些接口
- POSIX 信号量接口:
- 其他相关命令:
- 基于循环队列的生产者和消费者模型
- 同步关系
- 多生产多消费
我们今天接着来学习信号量:
什么是信号量
信号量(Semaphore)是一种用于操作系统中管理共享资源访问和同步的机制。它是一种特殊的数据结构,用来控制多个进程或线程对公共资源的访问,以防止多个进程同时对同一资源进行访问而导致的冲突问题。信号量维护了一个计数器,该计数器可以增加(通常称为V操作或Signal操作)或减少(称为P操作或Wait操作),并且这些操作都是原子的,即不可中断。
P操作(Wait操作)
- 当一个进程想要访问一个受保护的资源时,它会执行P操作。
- P操作会检查信号量的值,如果信号量大于0,则减1,并允许进程继续执行。
- 如果信号量等于0,表示资源已被占用,进程将被阻塞(等待)直到信号量的值变为非零。
V操作(Signal操作)
- 当一个进程完成对资源的访问后,它会执行V操作。
- V操作会将信号量的值加1,表示释放了一个资源。
- 如果有其他进程因为之前P操作而等待,此时可能会唤醒其中一个等待的进程。
信号量的类型
- 二值信号量:这种信号量只有0和1两种状态,
相当于一个互斥锁,用于实现互斥访问
。- 计数信号量:
可以用于控制有限数量的相同资源的访问
,计数器的初始值代表资源的数量。
简单来说,信号量的本质就是一个容量为N的锁,跟一般的锁不一样,它可以放个多个线程访问临界资源,但是达到上限就不会让线程进入了,而让他们阻塞等待。
一些接口
在Linux环境下,信号量作为进程间通信的一种手段,主要用于同步和互斥控制。以下是Linux下信号量的一些常用接口,主要涉及System V信号量和POSIX信号量两种类型:
POSIX 信号量接口:
- sem_init(sem_t *sem, int pshared, unsigned int value):
- 初始化一个POSIX信号量。
sem
是信号量的地址,pshared
指定信号量是否可以在多进程中共享(如果为1则是共享的),value
是信号量的初始值。
- sem_wait(sem_t *sem):[
- 执行P操作,如果信号量的值大于0,则减1并继续执行;否则,进程将被阻塞直到信号量的值大于0。
- sem_post(sem_t *sem):
- 执行V操作,增加信号量的值,如果因此唤醒了等待的进程,则会选择一个进行唤醒。
- sem_destroy(sem_t *sem):
- 销毁一个POSIX信号量。
- sem_getvalue(sem_t *sem, int *sval):
- 获取POSIX信号量的当前值,
sval
是存储信号量值的指针。
其他相关命令:
- ipcs: 查看系统中所有的IPC设施状态,包括消息队列、共享内存段和信号量。
- ipcrm: 删除指定的IPC设施,比如信号量集。
基于循环队列的生产者和消费者模型
我们这里模拟实现一个基于循环队列的生产者和消费者模型,首先我们实现一个循环队列:
先把架子搭好:
#include<iostream>
#include<semaphore>
#include<pthread.h>
#include<cstring>
#include<vector>
#include<semaphore.h>
#include<unistd.h>
template<class T>
class CircleQueue
{
public:
CircleQueue()
:_size(10)
,_product_start(0)
,_consum_start(0)
{
}
~CircleQueue()
{
}
private:
};
现在,我们要分析一下,这里面的同步关系:
同步关系
我们这里清楚,生产者会消费一个空间生产一个产品,并往前走一步,并且生产者和消费者是指向一个空间:
消费者会消费一个产品,腾出一个空间资源,然后往前走一步:
这个时候,我们可以分析出以下几条信息:
- 生产者一定先跑,因为一开始有空间资源,没有产品资源。
- 消费者一定比生产者跑的慢,因为消费者的速度是受生产者产出产品的速度决定的。
- 当空间资源被用完时,生产者停止生产,让消费者消费之后,腾出空间资源之后再继续生产。相反,如果没有产品资源,消费者阻塞,让生产者产出产品之后,再消费产品。
#include <iostream>
#include <semaphore>
#include <pthread.h>
#include <cstring>
#include <vector>
#include <semaphore.h>
#include <unistd.h>
// 定义一个泛型循环队列类,利用信号量实现线程安全的生产者消费者模型
template <class T>
class CircleQueue {
public:
// 默认构造函数,初始化一个大小为10的循环队列
CircleQueue()
: _size(10), // 队列默认容量
_product_start(0), // 生产者开始位置
_consum_start(0) // 消费者开始位置
{
// 初始化空间信号量,初始值为队列大小,表示可用空间数量
sem_init(&_sem_space, 0, _size);
// 初始化数据信号量,初始值为0,表示当前没有可消费的数据
sem_init(&_sem_data, 0, 0);
// 初始化队列容器
_queue.resize(_size);
}
// 带参数构造函数,允许用户自定义队列大小
CircleQueue(int size)
: _size(size), // 用户指定的队列容量
_product_start(0),
_consum_start(0)
{
sem_init(&_sem_space, 0, size); // 根据用户指定的大小初始化空间信号量
sem_init(&_sem_data, 0, 0);
_queue.resize(size);
}
// 析构函数,释放信号量资源
~CircleQueue() {
sem_destroy(&_sem_space);
sem_destroy(&_sem_data);
}
// 生产者方法,向队列中添加数据
void Push(const T& data) {
// 在尝试放入数据前,先等待确保有空闲空间
sem_wait(&_sem_space);
// 将数据放入队列的下一个生产位置
_queue[_product_start] = data;
// 更新生产者位置,并对索引取模以实现循环
_product_start = (_product_start + 1) % _size;
// 数据放入后,释放数据信号量,通知消费者有新数据可取
sem_post(&_sem_data);
}
// 消费者方法,从队列中取出数据
void Pop(T* out) {
// 等待直到有数据可消费
sem_wait(&_sem_data);
// 从队列的下一个消费位置取出数据
*out = _queue[_consum_start];
// 更新消费者位置,并对索引取模实现循环
_consum_start = (_consum_start + 1) % _size;
// 数据取出后,释放空间信号量,表明队列中有更多空间可填充
sem_post(&_sem_space);
}
private:
// 循环队列的底层数据结构
std::vector<T> _queue;
int _size; // 队列的最大容量
// 生产者和消费者的当前位置索引
int _product_start;
int _consum_start;
// 信号量用于同步控制
sem_t _sem_space; // 控制队列中的空闲空间
sem_t _sem_data; // 控制队列中的有效数据量
};
这段代码实现了一个基于信号量的线程安全循环队列模板类。它支持生产者线程向队列中添加元素(通过Push
方法),同时允许消费者线程从队列中取出元素(通过Pop
方法)。通过使用两个信号量——_sem_space
和 _sem_data
——分别管理队列的可用空间和有效数据量,确保了多线程环境下的正确同步与互斥。
基于这个,我们实现一下整体的代码:
#include"CircleQueue.hpp"
#include<time.h>
void* product(void* args)
{
CircleQueue<int>* cq = static_cast<CircleQueue<int>*>(args);
while(true)
{
//生产数据
int randomdata = rand() % 10 + 1;
cq->Push(randomdata);
std::cout << "Producter has product a number: " << randomdata <<
std::endl;
}
return nullptr;
}
void* consum(void* args)
{
CircleQueue<int>* cq = static_cast<CircleQueue<int>*>(args);
while(true)
{
//拿出数据
int outnumber = 0;
cq->Pop(&outnumber);
std::cout << "Consum gets a number: " << outnumber <<
std::endl;
sleep(1);
}
return nullptr;
}
int main()
{
srand(time(0));
//创建线程
pthread_t tid_product,tid_consum;
CircleQueue<int>*cq = new CircleQueue<int>();
//生产者
pthread_create(&tid_product,nullptr,product,cq);
//消费者
pthread_create(&tid_consum,nullptr,consum,cq);
pthread_join(tid_product,nullptr);
pthread_join(tid_consum,nullptr);
}
大家可以调整一下生产者或者消费者的速度,看看情况怎么样。
多生产多消费
这里注意一下,这里和互斥锁的情况有点不一样:
我们用一个通俗的例子来解释:
假设有一天,你和你的好朋友(一共8个人),想去电影院看电影:
你们到售票机哪里去买票,此时电影院的座位很充足,所以你们都买到票了。
但是一看座位号,发现大家全都是1号座位
这就很尴尬了,这个场景可以类比到我们上面的代码中,8个线程通过了信号量,但是都在往一个位置位置放东西,这样不行,所以我们得出位置是每个人独有的,一人一份,如果自己拥有,别人就不能拥有,所以为了保证每一个位置为一人独有所以我们要给每个位置上锁(电影院的座位)
如果有点绕,咋们来复盘一下:
- 电影院有很多位置,所以,我们多人可以都得到属于自己的位置(类比我们的循环队列)
- 为了保证我们的位置是独一无二属于自己,我们要给自己的位置上锁,保证只有自己可以坐这个座位。(类比循环队列中的下标)
解决完上面的问题,我们现在要做的,就是还要两把锁,一个保证生产的时候,放入时候的位子只属于一个生产者进程,另一把锁保证从一个位置里面拿产品的时候只属于一个消费者进程:
#include <iostream>
#include <semaphore>
#include <pthread.h>
#include <cstring>
#include <vector>
#include <semaphore.h>
#include <unistd.h>
// 定义一个泛型循环队列类,结合信号量与互斥锁实现线程安全的生产者消费者模型
template <class T>
class CircleQueue {
public:
// 默认构造函数,初始化一个大小为10的循环队列,并初始化信号量与互斥锁
CircleQueue()
: _size(10), // 队列默认容量
_product_start(0), // 生产者开始位置
_consum_start(0) // 消费者开始位置
{
sem_init(&_sem_space, 0, _size); // 初始化空间信号量,初始值为队列大小
sem_init(&_sem_data, 0, 0); // 初始化数据信号量,初始值为0
pthread_mutex_init(&_p_mutex, nullptr); // 初始化生产者互斥锁
pthread_mutex_init(&_c_mutex, nullptr); // 初始化消费者互斥锁
_queue.resize(_size); // 初始化队列向量
}
// 带参数构造函数,允许自定义队列大小
CircleQueue(int size)
: _size(size),
_product_start(0),
_consum_start(0)
{
sem_init(&_sem_space, 0, size);
sem_init(&_sem_data, 0, 0);
pthread_mutex_init(&_p_mutex, nullptr);
pthread_mutex_init(&_c_mutex, nullptr);
_queue.resize(size);
}
// 析构函数,释放信号量与互斥锁资源
~CircleQueue() {
sem_destroy(&_sem_space);
sem_destroy(&_sem_data);
pthread_mutex_destroy(&_p_mutex);
pthread_mutex_destroy(&_c_mutex);
}
// 生产者方法,向队列中添加数据
void Push(const T& data) {
// 确保有足够的空间再进行生产
sem_wait(&_sem_space);
// 使用互斥锁保护生产过程,防止与其它生产者并发冲突
pthread_mutex_lock(&_p_mutex);
_queue[_product_start] = data; // 添加数据
_product_start = (_product_start + 1) % _size; // 更新生产者位置
pthread_mutex_unlock(&_p_mutex); // 释放锁
sem_post(&_sem_data); // 数据添加完毕,释放数据信号量
}
// 消费者方法,从队列中取出数据
void Pop(T* out) {
// 确保有数据可消费
sem_wait(&_sem_data);
// 使用互斥锁保护消费过程
pthread_mutex_lock(&_c_mutex);
*out = _queue[_consum_start]; // 取出数据
_consum_start = (_consum_start + 1) % _size; // 更新消费者位置
pthread_mutex_unlock(&_c_mutex); // 释放锁
sem_post(&_sem_space); // 释放空间信号量
}
private:
// 循环队列的底层数据结构
std::vector<T> _queue;
int _size; // 队列的最大容量
// 生产者和消费者的当前位置索引
int _product_start;
int _consum_start;
// 同步控制工具
sem_t _sem_space; // 控制队列中的空闲空间
sem_t _sem_data; // 控制队列中的有效数据量
// 互斥锁用于保护队列访问的原子性
pthread_mutex_t _p_mutex; // 生产者使用的互斥锁
pthread_mutex_t _c_mutex; // 消费者使用的互斥锁
};
这里打印的时候,由于屏幕也是公共资源,我这里加锁,是保证打印的时候,只有生产者或者消费者打印信息:
#include"CircleQueue.hpp"
#include<time.h>
pthread_mutex_t global_mutex = PTHREAD_MUTEX_INITIALIZER;
void* product(void* args)
{
CircleQueue<int>* cq = static_cast<CircleQueue<int>*>(args);
while(true)
{
//生产数据
int randomdata = rand() % 10 + 1;
cq->Push(randomdata);
pthread_mutex_lock(&global_mutex);
std::cout << "Producter has product a number: " << randomdata <<
std::endl;
pthread_mutex_unlock(&global_mutex);
}
return nullptr;
}
void* consum(void* args)
{
CircleQueue<int>* cq = static_cast<CircleQueue<int>*>(args);
while(true)
{
//拿出数据
int outnumber = 0;
cq->Pop(&outnumber);
pthread_mutex_lock(&global_mutex);
std::cout << "Consum gets a number: " << outnumber <<
std::endl;
pthread_mutex_unlock(&global_mutex);
sleep(1);
}
return nullptr;
}
int main()
{
srand(time(0));
//创建线程
pthread_t tid_product[8],tid_consum[8];
CircleQueue<int>*cq = new CircleQueue<int>();
//生产者
for(int i = 0; i < 8; i++)
{
pthread_create(&tid_product[i],nullptr,product,cq);
}
//消费者
for(int i = 0; i < 8; i++)
{
pthread_create(&tid_consum[i],nullptr,consum,cq);
}
for(int i = 0; i < 8; i++)
{
pthread_join(tid_product[i],nullptr);
}
for(int i = 0; i < 8; i++)
{
pthread_join(tid_consum[i],nullptr);
}
}