【Linux】POSIX信号量与、基于环形队列实现的生产者消费者模型

一、POSIX信号量概述

信号量的基本概念

信号量在临界区的作用

与互斥锁的比较

信号量的原理

信号量的优势

二、信号量的操作

1、初始化信号量：sem_init

2、信号量申请（P操作）：sem_wait

3、信号量的释放（V操作）：sem_post

4、信号量的销毁：sem_destroy

信号量使用代码示例：

三、基于环形队列实现的生产者消费者模型

四、完整代码

五、生产者消费者模型两种实现方法的对比

阻塞队列与条件变量的模型

环形队列与信号量的模型

为什么环形队列与信号量需要两个锁

总结

一、POSIX信号量概述

POSIX信号量是POSIX标准（Portable Operating System Interface，可移植操作系统接口）定义的一种进程/线程间同步机制。它提供了一种跨平台的方法来实现进程间的同步和互斥。POSIX信号量可以用于控制对共享资源的访问，确保在任何时刻只有一个进程可以访问该资源，或者限制对资源的并发访问数量。

信号量的基本概念

信号量定义：
- 信号量是一个计数器，用于管理对有限资源的访问。
- 它通过维护一个计数值来表示资源的数量或可用的资源数。
操作：
- P操作（等待/减少操作）：当一个线程试图获取资源时，它会执行P操作（也叫sem_wait），这会减少信号量的计数。如果计数值已经为零，线程会被阻塞，直到有资源可用。
- V操作（信号/增加操作）：当一个线程释放资源时，它会执行V操作（也叫sem_post），这会增加信号量的计数。如果有线程在等待资源，这会唤醒其中一个线程。
- 阻塞与唤醒：POSIX 信号量通常有一个等待队列。当信号量的值为 0 或者低于某个阈值时，进程（或线程）在请求信号量时会被阻塞，并被加入到信号量的等待队列中。信号量操作系统内部会维护这个队列，以确保进程可以在信号量的值变为正数时被唤醒。

信号量在临界区的作用

控制访问：
在进入临界区之前，线程首先会通过执行P操作来申请信号量。这种操作检查信号量的计数，如果计数为零，线程会被阻塞，直到其他线程释放资源。
计数器机制：
信号量的计数器表示可用的资源数。当计数器大于零时，表示有足够的资源可用；当计数器为零时，表示资源已满或不可用，线程需要等待。
资源预定：
线程在进入临界区之前，先进行信号量的P操作（申请资源）。这种机制确保了临界区中的资源不会被过多的线程访问，从而避免了竞态条件和资源冲突。
退出临界区：
在离开临界区时，线程会执行V操作（释放资源）。这会增加信号量的计数，并可能唤醒等待资源的其他线程。

【注意：P和V操作必须保证是原子性的，这意味着在执行PV操作时，不会有其他线程干扰。】

与互斥锁的比较

互斥锁：主要用于保护临界区，确保同一时刻只有一个线程能访问临界区。互斥锁不管理资源的计数，仅用于保证互斥。
信号量：不仅提供互斥保护，还可以管理资源的数量。信号量适用于需要控制多个线程对有限资源访问的场景。

信号量的原理

我们将可能会被多个执行流同时访问的资源叫做临界资源，临界资源需要进行保护否则会出现数据不一致等问题。
当我们仅用一个互斥锁对临界资源进行保护时，相当于我们将这块临界资源看作一个整体，同一时刻只允许一个执行流对这块临界资源进行访问。
但实际我们可以将这块临界资源再分割为多个区域，当多个执行流需要访问临界资源时，如果这些执行流访问的是临界资源的不同区域，那么我们可以让这些执行流同时访问临界资源的不同区域，此时不会出现数据不一致等问题。

信号量的优势

信号量（信号灯）本质是一个计数器，是描述临界资源中资源数目的计数器，信号量能够更细粒度的对临界资源进行管理。

在基于阻塞队列的生产者消费者模型章节中，我们在访问阻塞队列时，同一时刻只会有一个线程成功竞争到互斥锁然后进入临界区当中。而只有进入临界区后，该执行流才能去判断阻塞队列中有多少剩余空间或剩余数据。因此我们不仅需要对阻塞队列中的数据条件进行循环判断，还需要在阻塞队列为空或为满时在条件变量处进行等待。

而使用信号量其实是对临界资源的预定机制。在进入临界区之前，每个线程都会先去申请临界区资源。这就使得线程不需要进入临界区，就可以预先完成临界资源的分配！同时，信号量也使线程对临界资源的操作更细致、更高效！

通过维护计数器，信号量可以准确地反映当前可用的资源数量。这使得线程在请求和释放资源时，能有效地控制资源的使用情况，避免了资源过度使用或饥饿状态。信号量通过原子操作来管理资源的计数，这可以减少竞争条件和锁争用，尤其是在高并发的环境中。

总的来说，信号量作为对临界资源的预定机制，通过精确控制资源的分配和使用，可以提高多线程程序的效率和稳定性。

二、信号量的操作

1、初始化信号量：sem_init

sem_init 是 POSIX 标准中用于初始化信号量的函数。它用于设置一个信号量对象的初始状态，包括其初始计数值和属性设置。

函数原型
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数说明

sem (sem_t *sem)：

这是指向一个 sem_t 类型的信号量对象的指针。这个对象在 sem_init 调用之前需要被分配空间（通常是通过静态分配或动态分配内存）。

pshared (int pshared)：

这个参数决定了信号量的共享范围。
如果 pshared 为 0，则信号量是线程间共享的，即同一个进程内的多个线程可以访问它。
如果 pshared 非 0，则信号量是进程间共享的，多个进程可以访问这个信号量。此时，sem 指向的信号量对象必须存储在共享内存区域中（例如，通过 mmap 分配的共享内存）。

value (unsigned int value)：

这是信号量的初始值，即信号量计数器的初始值。它表示资源的初始数量或可用数量。

返回值

成功：返回 0。
失败：返回 -1，并设置 errno 以指示错误类型。常见的错误包括 EINVAL（参数无效）或 ENOMEM（内存不足）。

2、信号量申请（P操作）：sem_wait

sem_wait 是 POSIX 标准中用于执行信号量的等待操作的函数。它用于请求信号量，通常在多线程或多进程环境中用于实现同步和互斥。

函数原型
int sem_wait(sem_t *sem);
参数说明

sem (sem_t *sem)：
这是指向一个已经初始化的信号量对象的指针。信号量对象必须在调用 sem_wait 之前通过 sem_init 或其他方式初始化。

返回值

成功：返回 0。
失败：返回 -1，并设置 errno 以指示错误类型。常见的错误包括 EINTR（操作被信号中断）或 EINVAL（无效的信号量描述符）。

功能和行为

等待信号量：sem_wait 函数会尝试将信号量的计数器减少 1。这个操作是原子性的，即在执行期间不会被中断。

阻塞和挂起：

如果信号量的计数器大于 0，sem_wait 会将计数器减 1，并立即返回，允许线程继续执行。
如果信号量的计数器为 0，调用 sem_wait 的线程会被阻塞，直到信号量的计数器变得大于 0。此时，线程会被唤醒，并且信号量计数器会减少 1。

原子性：sem_wait 的操作是原子性的，这意味着在 sem_wait 执行过程中，不会有其他线程或进程修改同一个信号量的状态，从而避免竞态条件。

3、信号量的释放（V操作）：sem_post

sem_post 是 POSIX 标准中用于释放信号量的函数。它的作用是将信号量的计数器增加 1，从而可能唤醒因等待信号量而被阻塞的线程或进程。

函数原型
int sem_post(sem_t *sem);
参数说明

sem (sem_t *sem)：
这是指向一个已经初始化的信号量对象的指针。信号量对象必须在调用 sem_post 之前通过 sem_init 或其他方式初始化。

返回值

成功：返回 0。
失败：返回 -1，并设置 errno 以指示错误类型。常见的错误包括 EINVAL（无效的信号量描述符）。

功能和行为

释放信号量：

sem_post 函数将信号量的计数器增加 1。这个操作是原子性的，即在执行期间不会被中断。

唤醒等待线程：

如果有线程因调用 sem_wait 而被阻塞（即信号量计数器为 0），sem_post 会增加信号量的计数器，并唤醒其中一个被阻塞的线程或进程，使其可以继续执行。

计数器上限：

在多数实现中，信号量的计数器没有上限，但实际限制取决于系统实现。在某些实现中，信号量计数器的最大值可能受到系统限制，因此要注意避免计数器溢出。

注意：sem_post 通常与 sem_wait 配对使用，确保在释放信号量时之前已经有线程因 sem_wait 被阻塞。否则，如果没有线程在等待信号量，sem_post 将增加计数器但不会产生实际的同步效果。

4、信号量的销毁：sem_destroy

sem_destroy 是 POSIX 标准中用于销毁信号量的函数。它的作用是释放与信号量对象相关的资源。在多线程或多进程编程中，使用信号量后，应确保在不再需要它们时调用 sem_destroy 来避免资源泄漏。

函数原型
int sem_destroy(sem_t *sem);
参数说明

sem (sem_t *sem)：
这是指向一个已经初始化的信号量对象的指针。该信号量必须在调用 sem_destroy 之前通过 sem_init 进行初始化，并且在调用 sem_destroy 时不应有线程正在等待或使用该信号量。

返回值

成功：返回 0。
失败：返回 -1，并设置 errno 以指示错误类型。常见的错误包括：
EINVAL：信号量不是有效的信号量或未被初始化。
EBUSY：信号量正在被其他线程使用（即，已经有线程在等待或已经获取了该信号量）。

功能和行为

释放资源：

sem_destroy 会释放与信号量相关的系统资源。这是重要的一步，尤其在使用动态分配的信号量时，以防止内存泄漏。

适用场景：

在信号量不再需要使用时调用，例如在程序结束前或线程完成工作后。

注意事项

调用时机：
只有在所有与信号量相关的线程已退出或不再使用该信号量时，才能成功地调用 sem_destroy。如果有线程仍在等待该信号量，调用将失败。

避免重复销毁：
不应多次调用 sem_destroy 来销毁同一个信号量，这可能导致未定义的行为。

初始化和销毁配对：
每个通过 sem_init 初始化的信号量都应以 sem_destroy 完成销毁，以确保资源的正确管理。

信号量使用代码示例：

#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <pthread.h>
#include <errno.h>

sem_t sem;

void* worker(void* arg) {
    // 等待信号量
    if (sem_wait(&sem) != 0) {
        perror("sem_wait failed");
        pthread_exit(NULL);
    }

    // 进入临界区
    printf("Thread %ld acquired the semaphore.\n", (long)arg);

    // 模拟工作
    sleep(2);

    // 释放信号量
    if (sem_post(&sem) != 0) {
        perror("sem_post failed");
        pthread_exit(NULL);
    }

    printf("Thread %ld released the semaphore.\n", (long)arg);

    pthread_exit(NULL);
}

int main() {
    pthread_t threads[2];
    int i;

    // 初始化信号量，初始值为 1
    if (sem_init(&sem, 0, 1) != 0) {
        perror("sem_init failed");
        exit(EXIT_FAILURE);
    }

    // 创建线程
    for (i = 0; i < 2; i++) {
        pthread_create(&threads[i], NULL, worker, (void*)(long)i);
    }

    // 等待线程完成
    for (i = 0; i < 2; i++) {
        pthread_join(threads[i], NULL);
    }

    // 销毁信号量
    sem_destroy(&sem);

    return 0;
}

三、基于环形队列实现的生产者消费者模型

对于环形队列，此前已经在数据结构中做过详细讲解，大家可移步观看。

在数据结构章节中，我们是如何判断环形队列是否为空或者为满呢？——是通过队首坐标和队尾坐标是否相差“1”来判断的。这也就意味着在这种做法下我们需要牺牲一个空间。

实际上，还有一种做法可以对环形队列的状态进行判断——使用一个计数器来实时记录环形队列中元素的个数。当像队列中添加数据时，计数器 +1；当队列弹出数据时，计数器 -1。当计数器为0或为我们所设定的队列最大容量时，代表环形队列为空或已满。

如此，我们就不必使用队首和队尾索引间的关系来对队列的状态进行判断。

而信号量的本质就是计数器，自然也能做上述工作！同时，信号量还可以承担为线程分配资源的功能！

生产者如何向环形队列中添加数据，消费者又如何从队列中提取数据呢？

队列的原则是“先进先出”，当我们向队列中添加元素时，应该从队尾开始添加。而提取元素时，应从队首开始提取。

我们使用数组模拟环形队列，由消费者代码控制队首索引：

//生产者承担添加元素的操作，形参为需要添加的对象的引用
    void Push(const T& in)
    {
        //1、申请一个代表空间资源的信号量，该信号量的值 -1 —— P操作
        P(&_space_nums);

        //2、去竞争临界区的互斥锁
        pthread_mutex_lock(&_productor_mutex);

        //3、向队尾添加元素，并让队尾索引指向下一个位置
        _ring_queue[_end] = in;
        _end = (_end + 1) % _max_capacity;

        //4、解锁
        pthread_mutex_unlock(&_productor_mutex);

        //5、此时生产者已经向环形队列中添加了一个数据
        //这意味着空间资源数量 -1， 数据资源数量 +1，此时对代表数据资源的信号量进行 +1 操作 —— V操作
        V(&_data_nums);
    }

由生产者代码来控制队尾索引：

//消费者承担提取元素的任务，形参为指向所提取的元素的指针，是输出型参数
    void Pop(T* out)
    {
        //1、申请一个代表数据资源的信号量，该信号量的值 -1 —— P操作
        P(&_data_nums);

        //2、去竞争临界区的互斥锁
        pthread_mutex_lock(&_consumer_mutex);

        //3、从队首提取元素，并让队首索引指向下一个位置
        *out = _ring_queue[_front];
        _front = (_front + 1) % _max_capacity;

        //4、解锁
        pthread_mutex_unlock(&_consumer_mutex);

        //5、此时消费者已经从环形队列中提取了一个数据
        //这意味着数据资源数量 -1， 空间资源数量 +1，此时对代表空间资源的信号量进行 +1 操作 —— V操作
        V(&_space_nums);
    }

生产者关注的是空间资源，消费者关注的是数据资源

对于生产者和消费者来说，它们关注的资源是不同的：

生产者关注的是环形队列当中是否有空间（blank），只要有空间生产者就可以进行生产。
消费者关注的是环形队列当中是否有数据（data），只要有数据消费者就可以进行消费。

blank_sem和data_sem的初始值设置

现在我们用信号量来描述环形队列当中的空间资源（blank_sem）和数据资源（data_sem），在我们初始信号量时给它们设置的初始值是不同的：

blank_sem的初始值我们应该设置为环形队列的容量，因为刚开始时环形队列当中全是空间。
data_sem的初始值我们应该设置为0，因为刚开始时环形队列当中没有数据。

template <typename T>
class RingQueue
{
private:
    sem_t _data_nums;//数组资源信号量
    sem_t _space_nums;//空间资源信号量
    pthread_mutex_t _productor_mutex;//生产操作临界区的互斥锁
    pthread_mutex_t _consumer_mutex;//消费操作临界区的互斥锁
    int _max_capacity; //环形队列的最大容量
    std::vector<T> _ring_queue;//数组模拟环形队列
    int _front;//队首索引，由消费者来维护
    int _end;//队尾索引，由生产者来维护
public:
    RingQueue(int max_capicity = MAXCAPACITY)
    :_max_capacity(max_capicity), _ring_queue(max_capicity)
    {
        _front = _end = 0;
        sem_init(&_data_nums, 0, 0);//数据资源信号量初始化为0
        sem_init(&_space_nums, 0, _max_capacity);//空间资源信号量初始化为环形队列的最大容量
        pthread_mutex_init(&_productor_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }
}

生产者申请空间资源，释放数据资源

对于生产者来说，生产者每次生产数据前都需要先申请blank_sem：

如果blank_sem的值不为0，则信号量申请成功，此时生产者可以进行生产操作。
如果blank_sem的值为0，则信号量申请失败，此时生产者需要在blank_sem的等待队列下进行阻塞等待，直到环形队列当中有新的空间后再被唤醒。

当生产者生产完数据后，应该释放data_sem：

虽然生产者在进行生产前是对blank_sem进行的P操作，但是当生产者生产完数据，应该对data_sem进行V操作而不是blank_sem。
生产者在生产数据前申请到的是blank位置，当生产者生产完数据后，该位置当中存储的是生产者生产的数据，在该数据被消费者消费之前，该位置不再是blank位置，而应该是data位置。
当生产者生产完数据后，意味着环形队列当中多了一个data位置，因此我们应该对data_sem进行V操作。

消费者申请数据资源，释放空间资源

对于消费者来说，消费者每次消费数据前都需要先申请data_sem：

如果data_sem的值不为0，则信号量申请成功，此时消费者可以进行消费操作。
如果data_sem的值为0，则信号量申请失败，此时消费者需要在data_sem的等待队列下进行阻塞等待，直到环形队列当中有新的数据后再被唤醒。

当消费者消费完数据后，应该释放blank_sem：

虽然消费者在进行消费前是对data_sem进行的P操作，但是当消费者消费完数据，应该对blank_sem进行V操作而不是data_sem。
消费者在消费数据前申请到的是data位置，当消费者消费完数据后，该位置当中的数据已经被消费过了，再次被消费就没有意义了，为了让生产者后续可以在该位置生产新的数据，我们应该将该位置算作blank位置，而不是data位置。
当消费者消费完数据后，意味着环形队列当中多了一个blank位置，因此我们应该对blank_sem进行V操作。

必须遵守的两个规则

在基于环形队列的生产者和消费者模型当中，生产者和消费者必须遵守如下两个规则。

第一个规则：生产者和消费者不能对同一个位置进行访问。

生产者和消费者在访问环形队列时：

如果生产者和消费者访问的是环形队列当中的同一个位置，那么此时生产者和消费者就相当于同时对这一块临界资源进行了访问，这当然是不允许的。
而如果生产者和消费者访问的是环形队列当中的不同位置，那么此时生产者和消费者是可以同时进行生产和消费的，此时不会出现数据不一致等问题。

如下图：

第二个规则：无论是生产者还是消费者，都不应该将对方套一个圈以上。

生产者从消费者的位置开始一直按顺时针方向进行生产，如果生产者生产的速度比消费者消费的速度快，那么当生产者绕着消费者生产了一圈数据后再次遇到消费者，此时生产者就不应该再继续生产了，因为再生产就会覆盖还未被消费者消费的数据。
同理，消费者从生产者的位置开始一直按顺时针方向进行消费，如果消费者消费的速度比生产者生产的速度快，那么当消费者绕着生产者消费了一圈数据后再次遇到生产者，此时消费者就不应该再继续消费了，因为再消费就会消费到缓冲区中保存的废弃数据。

如下图：

四、完整代码

#include <vector>
#include <pthread.h>
#include <semaphore.h>

const int MAXCAPACITY = 6; 

template <typename T>
class RingQueue
{
private:
    sem_t _data_nums;//数组资源信号量
    sem_t _space_nums;//空间资源信号量
    pthread_mutex_t _productor_mutex;//生产操作临界区的互斥锁
    pthread_mutex_t _consumer_mutex;//消费操作临界区的互斥锁
    int _max_capacity; //环形队列的最大容量
    std::vector<T> _ring_queue;//数组模拟环形队列
    int _front;//队首索引，由消费者来维护
    int _end;//队尾索引，由生产者来维护
public:
    RingQueue(int max_capicity = MAXCAPACITY)
    :_max_capacity(max_capicity), _ring_queue(max_capicity)
    {
        _front = _end = 0;
        sem_init(&_data_nums, 0, 0);//数据资源信号量初始化为0
        sem_init(&_space_nums, 0, _max_capacity);//空间资源信号量初始化为环形队列的最大容量
        pthread_mutex_init(&_productor_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }

    void P(sem_t* sem)
    {
        sem_wait(sem);
    }
    
    void V(sem_t* sem)
    {
        sem_post(sem);
    }

    //生产者承担添加元素的操作，形参为需要添加的对象的引用
    void Push(const T& in)
    {
        //1、申请一个代表空间资源的信号量，该信号量的值 -1 —— P操作
        P(&_space_nums);

        //2、去竞争临界区的互斥锁
        pthread_mutex_lock(&_productor_mutex);

        //3、向队尾添加元素，并让队尾索引指向下一个位置
        _ring_queue[_end] = in;
        _end = (_end + 1) % _max_capacity;

        //4、解锁
        pthread_mutex_unlock(&_productor_mutex);

        //5、此时生产者已经向环形队列中添加了一个数据
        //这意味着空间资源数量 -1， 数据资源数量 +1，此时对代表数据资源的信号量进行 +1 操作 —— V操作
        V(&_data_nums);
    }

    //消费者承担提取元素的任务，形参为指向所提取的元素的指针，是输出型参数
    void Pop(T* out)
    {
        //1、申请一个代表数据资源的信号量，该信号量的值 -1 —— P操作
        P(&_data_nums);

        //2、去竞争临界区的互斥锁
        pthread_mutex_lock(&_consumer_mutex);

        //3、从队首提取元素，并让队首索引指向下一个位置
        *out = _ring_queue[_front];
        _front = (_front + 1) % _max_capacity;

        //4、解锁
        pthread_mutex_unlock(&_consumer_mutex);

        //5、此时消费者已经从环形队列中提取了一个数据
        //这意味着数据资源数量 -1， 空间资源数量 +1，此时对代表空间资源的信号量进行 +1 操作 —— V操作
        V(&_space_nums);
    }

    ~RingQueue()
    {
        sem_destroy(&_data_nums);
        sem_destroy(&_space_nums);
        pthread_mutex_destroy(&_productor_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }
};

五、生产者消费者模型两种实现方法的对比

阻塞队列与条件变量的模型

在基于阻塞队列的实现中，生产者和消费者之间使用条件变量进行同步，而整个队列由一个互斥锁保护。这种方法的特点是将队列和同步机制视为一个整体进行管理：

单一互斥锁：使用一把锁来保护整个队列，确保生产者和消费者在访问队列时不会同时进行操作，避免数据竞争和不一致性。
条件变量：用来实现生产者和消费者之间的协调。生产者在队列满时等待，消费者在队列空时等待。条件变量的等待和通知都依赖于持有的互斥锁。

优点：

简单：由于只需要一把锁，代码实现和管理都相对简单。
一致性：锁保护了整个队列，确保了在任何时刻，队列的状态是一致的。

环形队列与信号量的模型

环形队列与信号量的实现方法通常涉及到两个独立的锁（或信号量），一个用于管理队列的访问，另一个用于管理队列的空闲和满状态：

两个独立的锁/信号量：
互斥锁：用于保护对队列的访问，确保生产者和消费者在对队列进行操作时是互斥的——即保护线程对临界资源操作的原子性。
信号量：用来表示队列中的空位和数据项数量。生产者和消费者通过信号量进行同步，但信号量与互斥锁是分开的。

优点：

细化的控制：信号量提供了对队列空闲和满状态的精细控制，使得生产者和消费者可以更高效地进行等待和通知，减少了不必要的等待时间。
独立管理：通过信号量，生产者和消费者在访问队列的逻辑上是分开的，互斥锁仅保护对队列的实际操作，而信号量负责同步和状态管理。

为什么环形队列与信号量需要两个锁

资源管理的独立性：
在环形队列的模型中，互斥锁保护队列的操作（如插入和删除），而信号量管理队列的状态（如队列空闲和满状态）。这两个机制的职责不同，因此需要独立的锁/信号量来实现更精细的控制。

提高效率：
使用信号量来管理空闲和满状态，可以减少生产者和消费者之间的等待时间。生产者和消费者可以在不同的信号量上进行操作，而不需要在操作队列时额外等待，从而提高了系统的效率。

避免复杂的锁竞争：
在使用信号量的情况下，生产者和消费者可以在互斥锁保护的范围之外进行状态检查和信号操作，从而减少了对互斥锁的竞争和锁持有的时间。

总结

阻塞队列与条件变量的模型：一个互斥锁保护整个队列，条件变量用于协调生产者和消费者。这种方法简单且一致性好，适合于简单的生产者消费者问题。

环形队列与信号量的模型：两个独立的锁/信号量分别管理队列的访问和状态。这种方法提供了更细粒度的控制，提高了效率，但实现和管理上相对复杂。