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

【多线程】七、POSIX信号量环形队列的生产者消费者模型

news 来源：原创 2025/7/17 5:57:50

文章目录

Ⅰ. 信号量
- 一、POSIX 信号量的概念
- 二、POSIX 信号量的类型区别
- 三、POSIX 信号量与 SystemV 信号量的区别
Ⅱ. 线程信号量基本原理
- 一、为什么要引入信号量❓
- 二、PV 操作
- 三、POSIX 信号量的实现原理
- 四、CAS操作介绍
Ⅲ. POSIX未命名信号量接口
- 一、初始化无名信号量
- 二、销毁无名信号量
- 三、阻塞减少信号量（相当于P操作）
- 四、非阻塞减少信号量（相当于P操作）
- 五、增加信号量（相当于V操作）
- 六、读取当前信号量值
- 测试代码：利用信号量实现互斥锁功能
Ⅳ. 基于环形队列的生产者消费者模型
- 一、环形队列的应用场景
- 二、基于环形队列的生产者消费者模型的实现
- - ① 无保存者版本
  - ② 有保存者版本
Ⅳ. 效率高在哪里呢❓❓❓

Ⅰ. 信号量

一、POSIX 信号量的概念

POSIX信号量是一种进程间通信机制，它允许进程在共享资源上进行同步和互斥访问。POSIX信号量是一种计数信号量，它可以被多个进程共享，并且可以通过系统调用进行控制。POSIX 信号量是一种强大的进程间通信机制，它可以帮助程序员实现复杂的同步和互斥操作，确保多个进程之间共享资源的正确性和安全性。

信号量通常用来协调对资源的访问，其中信号计数会初始化为可用资源的数目。然后线程在资源增加时会增加计数，在删除资源时会减小计数，这些操作都以原子方式执行的！如果信号计数变为零，则表明已无可用资源。计数为零时，尝试减小信号的线程会被阻塞，直到计数大于零为止。

POSIX 信号量具有以下特点：

可以被多个进程共享，或者被同一进程中的线程共享。
可以被用来进行同步和互斥访问共享资源。
具有 原子性的计数功能，允许多个进程或者线程同时访问资源。

信号量就好比电影院的票数，而我们去买票的操作就是下面会讲的 P 操作，退票的操作就是 V 操作，这些操作的过程中是原子性的，并且对于电影票数，我们是可以提前得知的，也就意味着我们不需要提前加锁去判断，因为对电影票的操作是原子性的，所以不需要关心线程安全的问题！其实这都利用了一些技术比如 CAS（Compare-And-Swap） 等，这些下面会介绍到！

二、POSIX 信号量的类型区别

POSIX信号量有两种类型：命名信号量和未命名信号量。它们有以下区别：

命名方式不同：命名信号量通过一个字符串名字来标识，可以被多个进程共享，而未命名信号量只能被同一进程内的线程共享，不需要名字。
创建方式不同：创建命名信号量时需要使用 sem_open() 函数，创建未命名信号量时需要使用 sem_init() 函数。
销毁方式不同：销毁命名信号量时需要使用 sem_unlink() 函数，销毁未命名信号量时需要使用 sem_destroy() 函数。
访问权限不同：命名信号量可以通过文件系统的权限机制来控制对其的访问权限，而对于未命名信号量，访问权限只能通过进程间的 UID 和 GID 来控制。
在系统中的存储位置不同：命名信号量被存储在文件系统中的一个特殊目录中，而未命名信号量被存储在进程的地址空间中。

其中最重要的区别就是 命名信号量通过一个字符串名字来标识，可以被多个进程共享。未命名信号量只能被同一进程内的线程共享。而接下来我们主要学习的是未命名信号量，因为其可以帮助我们实现同一进程内的线程共享，达到多线程编程目的！

三、POSIX 信号量与 SystemV 信号量的区别

POSIX 信号量和 SystemV 信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的，但POSIX可以用于线程间同步。除此之外，还有一些其它的区别，下面列举出来：

编程接口不同： POSIX 信号量使用函数 sem_open()、sem_close()、sem_wait()、sem_post() 等来实现，而 SystemV 信号量使用函数 semget()、semctl()、semop() 等来实现。
信号量命名方式不同： POSIX 信号量采用类似于文件路径的命名方式，而 SystemV 信号量采用整数 key 值的命名方式。
信号量数量不同： POSIX 信号量可以创建多个信号量，而 SystemV 信号量只有一组信号量。
处理进程终止的方式不同： POSIX 信号量可以通过 sem_close() 和 sem_unlink() 函数来释放信号量资源，而 SystemV 信号量需要使用 semctl() 函数来进行释放。
信号量操作的方式不同： POSIX 信号量的操作更加简洁明了，使用 sem_wait() 和 sem_post() 两个函数即可，而 SystemV 信号量需要通过 semop() 函数来实现信号量操作。

总之， POSIX 信号量和 SystemV 信号量在使用方式、命名方式、数量、释放方式和操作方式等方面存在一些不同，程序员在选择使用哪一种信号量时需要根据具体的情况来决定。

Ⅱ. 线程信号量基本原理

一、为什么要引入信号量❓

在引入一种新方法之前，肯定是因为我们之前使用的一些方法比如说条件变量与互斥量，存在一些短板，所以我们通过新方法可以解决这些存在的问题！

还记得我们之前学条件变量的时候，为了实现互斥和同步，我们需要这样子使用互斥量和条件变量：

// 伪代码：
pthread_mutex_lock(&mutex);
while(condition_is_false)
{pthread_cond_wait(&cond, &mutex);
}
// 操作临界资源
// 修改条件
pthread_mutex_unlock(&mutex);

上述代码中，我们每次要去操作临界资源的时候，必须先上锁，因为我们不知道条件变量是否满足，如果不提前上锁就去访问条件变量和临界资源的话，就有很大可能导致线程不安全的问题，而我们也无法在条件变量已有的资源情况下去解决这个问题。

所以对于条件变量的使用每次我们就得上锁，但是我们知道，上锁是有消耗的，每次我们要去判断条件是否成立之前，如果这个条件已经不成立了，但是我们还得先上锁才能得知不成立，这不是明显消耗了不必要的资源了吗。

所以我们就要学习信号量，信号量说简单一点也是一个共享资源，但是对它的操作，是原子性的，从而达到了不需要加锁就能实现判断是否需要满足信号量的要求从而对临界资源进行操作！

但我们要明白的是，其实互斥锁本质也是一种信号量，只不过它只有 0 和 1 两种状态，表示是否被锁定，所以我们是可以用信号量来代替互斥锁的！

二、PV 操作

PV 操作也称为 wait() 和 signal() 操作，是操作系统中用于操作信号量的基本操作，它们通常与信号量一起使用来进行进程或者线程间的同步和互斥访问。

PV 操作分为两种：

P操作（proberen荷兰语：尝试降低），即 wait() 操作：它用于获取（获取或占用）一个资源或者等待某个事件的发生，其作用是 将信号量的值减一，如果 信号量的值为0，则调用进程或者线程会阻塞，等待资源或事件的发生。
V操作（verhogen荷兰语：增长），即 signal() 操作：它用于释放一个资源或者发出某个事件，其作用是 将信号量的值加一，如果 有其他进程或者线程在等待该信号量，则唤醒其中一个进程，让它可以继续执行。

例如，当多个进程需要访问共享资源时，可以使用 PV操作 来实现互斥访问，其中 P操作 用于 占用资源，V操作 用于 释放资源。又例如，当多个进程需要等待某个事件的发生时，可以使用 PV操作 来实现等待和唤醒的操作，其中 P操作 用于等待事件的发生，V操作 用于发出事件的信号。

需要注意的是，在使用 PV 操作时，需要考虑信号量的初始值，以及多个进程同时访问时可能出现的竞争条件，从而保证操作的正确性和安全性。

三、POSIX 信号量的实现原理

POSIX信号量的实现通常是通过原子操作来保证多个进程之间对信号量的操作是原子性的。

具体地说，POSIX信号量的实现使用了一些底层硬件原语或操作系统提供的原子操作，比如 CAS（Compare-and-Swap） 等，以确保对信号量的操作是原子性的，从而避免了竞态条件和死锁等问题的出现。

原子操作是一种不可中断的操作，可以在多线程或多进程并发访问共享资源的情况下，保证对共享资源的访问是原子性的。因此，使用原子操作来实现信号量操作，可以有效地避免因多进程并发访问而导致的数据竞争和互斥访问等问题。

需要注意的是，原子操作在一些平台上的实现方式可能会有所不同，程序员在使用时需要根据具体的平台和操作系统来选择适当的原子操作实现。

四、CAS操作介绍

Compare-and-Swap（CAS）是一种常见的原子操作，也称为“比较-交换”操作，是一种比较后数据若无改变则交换数据的一种无锁操作(乐观锁)。用于实现多线程或多进程并发访问共享资源的原子性操作。

CAS 操作通常有三个参数：共享变量的内存地址、期望值和新值。CAS 操作首先比较共享变量的值是否等于期望值，如果相等，则将共享变量的值替换为新值；如果不相等，则不执行任何操作。在这个过程中，CAS操作具有原子性，即其他线程或进程无法同时访问共享变量，从而避免了竞态条件和数据竞争等问题。

CAS 操作常见于并发控制算法中，如锁、信号量、读写锁等，它可以用来保证对共享资源的访问是原子性的，从而实现并发控制的正确性和安全性。

需要注意的是，CAS 操作虽然可以保证对共享变量的原子性操作，但是由于 CAS 操作涉及到对共享变量的读取和修改，因此仍然可能存在ABA问题，即在多线程或多进程并发访问共享变量的情况下，共享变量的值在某个时刻先变成了 A，然后又变成了 B，最后再变回了 A，从而导致一些问题。

为了避免上述的问题，通常需要使用带有版本号等标识的 CAS 操作来解决。即在进行 CAS 操作时，不仅要比较共享变量的值是否等于期望值，还需要比较版本号是否匹配。这样可以确保即使共享变量的值在中间被修改过，但由于版本号不匹配，CAS 操作依然会失败，从而避免了上述问题的出现。

CAS的伪代码如下：

template bool CAS(T* addr, T expected, T value)
{if(*addr == expected){*addr=value;return true;}return false;
}
int count=0;
void count_atomic_inc(int*addr)
{int oldval = 0;int newval = 0;do{oldval = *addr;newval = oldval + 1;}until CAS(addr, oldval, newval)
}// 简化之后，CAS比较与交换的伪代码可以表示为：
do{备份旧数据；基于旧数据构造新数据；}while(!CAS( 内存地址，备份的旧数据，新数据 ))

Ⅲ. POSIX未命名信号量接口

我们主要是为了完成线程间的通信，所以下面主要介绍的是未命名信号量的接口，至于命名信号量接口，可以自行上网查阅！

一、初始化无名信号量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);// 功能：初始化一个无名信号量
// 返回值：成功返回0，失败返回-1，并设置errno变量表示出错的原因

其中参数：

sem：指向要初始化的信号量的指针。
pshared：0 表示线程间共享，此时为未命名信号量；非 0 表示进程间共享，此时为命名信号量。
value：信号量初始值，最大可以设置为 2147483647（SEM_VALUE_MAX）。

对于 sem_t 类型结构体的定义如下：

typedef struct
{struct _pthread_fastlock __sem_lock; // 用于保护信号量的互斥锁，是一个pthread_fastlock类型的对象。int __sem_value; // 表示信号量的当前值，即信号量中的可用资源数。_pthread_descr __sem_waiting; // 表示正在等待信号量的线程或进程的列表，是一个_pthread_descr类型的对象。
} sem_t;

二、销毁无名信号量

#include <semaphore.h>
int sem_destroy(sem_t *sem);// 功能：销毁一个无名信号量
// 返回值：成功返回0，失败返回-1，并设置errno变量表示出错的原因
// 参数：sem是指向要销毁的信号量的指针。

三、阻塞减少信号量（相当于P操作）

#include <semaphore.h>
int sem_wait(sem_t *sem);// 功能：阻塞减少信号量，直到 sem 所指示的信号量计数大于零为止，之后以原子方式减小计数。
// 返回值：成功返回0，失败返回-1，并设置errno变量表示出错的原因。
// 参数：sem是指向要阻塞减少的信号量的指针。

四、非阻塞减少信号量（相当于P操作）

#include <semaphore.h>
int sem_trywait(sem_t *sem);// 功能：在计数大于零时，尝试以原子方式减小sem所指示的信号量计数。
// 返回值：成功返回0，失败返回-1，并设置errno变量表示出错的原因。
// 参数：sem是指向要非阻塞减少的信号量的指针。

五、增加信号量（相当于V操作）

#include <semaphore.h>
int sem_post(sem_t *sem);// 功能：以原子方式增加sem所指示的信号量，或者说给信号量解锁。
// 返回值：成功返回0，失败返回-1，并设置errno变量表示出错的原因。
// 参数：sem是指向要增加的信号量的指针。

六、读取当前信号量值

#include <semaphore.h>
int sem_getvalue(sem_t *sem, int *sval);// 功能：读取当前信号量的值。
// 返回值：成功返回0，并将该信号量当前值存储在sval变量中；失败返回-1。
// 参数：
//     sem：指向要增加的信号量的指针。
//     sval：输出型参数，存放信号量值

测试代码：利用信号量实现互斥锁功能

#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>int global_value = 20;
sem_t mutex; // 使用信号量来实现代替互斥锁void* thread_routine(void* args)
{char* namebuffer = static_cast<char*>(args);while(global_value){sem_wait(&mutex); // 将mutex减一，变成0，相当于加锁，其它线程会阻塞if(global_value > 0)std::cout << namebuffer << " the global_value is " << global_value-- << std::endl;sem_post(&mutex); // 将mutex加一，变成1，相当于解锁，其它线程就能竞争该信号量usleep(10000);}return nullptr;
}int main()