【Linux】信号量到底在数什么：从PV操作到RingQueue环形队列生产者消费者模型

---

前言

上篇文章我们讲了用「互斥锁 + 条件变量」实现的阻塞队列。阻塞队列是把整个队列当成一个整体来使用的，而如果我们想把临界资源按块拆开、分批给不同线程使用，就需要另一种同步工具——信号量。

这一篇主要回答三个问题：

1. 信号量中的计数值到底表示什么；
2. 环形队列为什么需要“空格”和“数据”两个信号量；
3. 单生产单消费扩展到多生产多消费后，为什么还要增加两把互斥锁。

一、信号量

1.1 回顾相关概念

操作	别名	含义	伪代码
P	wait / sem_wait	申请资源，不足则阻塞	if(--sem<0) block
V	post / sem_post	释放资源并唤醒等待者	++sem; wake_one

📌 注意表格里的 if(--sem<0) block 是「教材抽象模型」：这是 Dijkstra 经典信号量教学模型，允许内部抽象计数出现负值来表示等待者数量。而下面 POSIX 的 sem_wait 要按实际接口语义理解——值为 0 时阻塞，不会让你看到负数。两者是不同抽象层次，不要拿负数模型去硬套 sem_t。

信号和信号量只是名字相近，解决的问题并不相同：信号用于异步通知，信号量用于资源计数和同步。

可以先把信号量理解成一个计数器。计数为 1 时，同一时刻最多允许一个线程成功申请资源，表现得像二元信号量；计数大于 1 时，表示当前还有多份资源可供申请。

还是用电影院理解：VIP 厅只有一个座位，计数初值就是 1；普通影厅有 N 个座位，计数初值就是 N。每进入一个人，可用座位减 1；有人离开，可用座位加 1。

前面写阻塞队列时，std::queue 被当成一个整体保护。一个线程执行 push，另一个线程执行 pop，都会读写同一个容器对象的内部状态。如果没有互斥锁，就会产生数据竞争，程序行为未定义。

多线程使用资源，有两种场景：

1. 将目标资源作为一个整体保护：通常使用互斥锁，条件不满足时再配合条件变量；
2. 将资源划分成多个可独立使用的“块”：使用信号量记录还剩多少份资源。

所有线程都会访问同一个信号量对象，所以计数值本身也必须被安全修改。申请资源对应 P 操作，释放资源对应 V 操作；这两个操作都要具有原子性，不能直接拿普通整数随意 -- 或 ++。

先记住：P 操作申请一份资源，V 操作归还或增加一份资源。资源不足时，P 操作会让调用线程等待。

1.2 用信号量实现环形队列生产者消费者模型

1.2.1 先解决环形队列的判空与判满

维度	信号量	互斥锁	条件变量
资源计数	有（N）	无（0/1）	无
谁释放	任意线程	持有者	配合 mutex 使用
典型用途	控制并发数 / 生产消费	保护临界区	等待某条件成立

资源被拆成多份后，还要解决三个问题：用什么结构保存资源、如何判断剩余数量、如何避免多个线程访问同一个位置。这里使用固定大小的数组模拟环形队列。

• 环形队列为空：head == tail
• 环形队列满的时候：head == tail

环形队列满和空都是 head == tail，怎么分清？

• 方案 1：额外维护计数器。入队时 count++，出队时 count--；count == 0 表示空，count == N 表示满。
• 方案 2：预留一个位置。head == tail 表示空，(tail + 1) % N == head 表示满。

环形队列的逻辑示意图：

第一张图中，head 与 tail 指向不同位置，说明队列中已经有一段连续的有效区间。随着插入和删除不断进行，两个下标都会向后移动；走到数组末尾后，再通过取模回到下标 0。

第二张图中，head 与 tail 又回到了同一个位置。只看这两个下标，既可能表示队列为空，也可能表示队列已经装满。因此普通环形队列还需要额外的计数器、标记位，或者预留一个空位置来区分空和满。

本文使用长度为 N 的数组。head 和 tail 从 0 开始，每次移动后都对 N 取模，这样下标到达数组末尾后会重新绕回 0。

上面便是数据结构中环形队列的介绍。而今天我们不再使用如上方法判断环形队列的空与满，我们可以使用信号量来判断满和空。

1.2.2 用“圆桌和盘子”理解生产消费关系

可以把环形队列想象成一张圆桌，每个格子都是一个盘子。生产者沿着 tail 放苹果，消费者沿着 head 取苹果，每个盘子只能放一个。

盘子为空时，消费者不能拿；盘子放满后，生产者不能继续覆盖原来的苹果。只有生产者和消费者没有访问同一个位置时，两边才可以同时进行。

• 规则 1：队列为空时没有数据，消费者必须等待生产者；
• 规则 2：队列已满时没有空格，生产者必须等待消费者；
• 规则 3：生产者不能覆盖尚未消费的数据；
• 规则 4：消费者不能读取尚未生产的位置。

只要生产者和消费者访问的不是同一个位置，两边就可以同时进行。例如生产者在第 6 个盘子放苹果时，消费者可以从第 1 个盘子取苹果。真正需要安排先后顺序的是空和满这两个边界状态。

• 队列为空：消费者等待，生产者先运行；
• 队列已满：生产者等待，消费者先运行；
• 队列既不空也不满：生产和消费可以并发进行。

生产者关心的是“还有多少空格”，消费者关心的是“还有多少份有效数据”。队列初始为空，所以空格资源为 N，数据资源为 0。

这一节先讨论单生产、单消费。多生产、多消费还要处理同类线程之间的竞争，后面的代码会补上互斥锁。

1.3 POSIX 信号量接口

1.3.1 用两个信号量描述空格和数据

生产者资源：sem_blank = N（空格），定义变量指向开始位置 p_step = 0；
消费者资源：sem_data = 0；定义变量指向开始位置 c_step = 0；

生产者写入数据前，先申请一个空格；写入完成后，再发布一份可消费的数据：

P(sem_blank);  // blank--
// 在 p_step 进行生产
p_step++;
p_step %= N;
V(sem_data);   // data++

消费者读取数据前，先申请一份有效数据；读取完成后，再归还一个空格：

P(sem_data);
// 在 c_step 的位置进行消费
c_step++;
c_step %= N;
V(sem_blank);

P 操作是原子的。申请成功，线程继续运行；资源为 0 时，调用线程等待，不会继续访问队列。

当空格为 N，数据为 0 时，是生产者先运行。因为为空的时候，环形队列没有数据，消费者的 P 操作会阻塞。

空格信号量减到 0 时，说明队列已经写满，生产者会停在下一次 P 操作。消费者取走一份数据后，通过 V 操作归还一个空格，生产者才有机会继续写入。

同理，数据信号量为 0 时，消费者必须等待生产者发布数据。两个信号量分别把“空”和“满”这两个边界条件表达清楚了。

1. empty 的资源 = 空闲盘子格子
   • P(empty)：拿走一个空盘子
   • V(empty)：归还腾空的盘子格子（真正释放盘子）
2. full 的资源 = 已经装好苹果的数据
   • P(full)：拿走一份苹果数据
   • V(full)：产出一份苹果数据（仅仅增加"可消费数据"，盘子格子不释放！）

V(full) 只是宣告多了一份可消费数据，不会释放盘子；只有消费者执行 V(empty)，才表示一个盘子重新空出来。

P 各自原子，V 各自原子；P 与 V 之间可以发生线程切换。

POSIX 和 System V 都能完成资源计数与同步，但接口形式和管理方式不同。本文使用的是 POSIX 无名信号量，它适合放在进程内做线程同步；如果用于进程间同步，信号量对象还必须位于共享内存中。

1.3.2 常用接口

初始化信号量

• pshared == 0：同一进程中的线程共享；
• pshared != 0：进程间共享，此时 sem_t 必须放在共享内存中；
• value：信号量初始值。

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

销毁信号量

int sem_destroy(sem_t *sem);

申请资源

int sem_wait(sem_t *sem); // P()

资源可用时，sem_wait 完成一次申请；计数为 0 时，调用线程等待。如果调用被信号中断，还应根据返回值和 errno == EINTR 决定是否重试。

发布资源

int sem_post(sem_t *sem); // V()

sem_post 增加一份可用资源，并可能唤醒一个等待者。这里的“资源”由程序自己定义：可以是空格，也可以是已经生产好的数据。

上一节生产者 - 消费者的例子是基于 queue 的，其空间可以动态分配，现在基于固定大小的环形队列重写这个程序（POSIX 信号量）：

1.4 代码实现

1.4.1 信号量的封装

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

namespace SemModule
{
    // 对 POSIX 信号量的极薄封装：把 P/V 语义直接暴露出来
    class Sem
    {
    public:
        Sem(unsigned int value)
        {
            // 第二个参数 0 表示信号量在「同一进程内的线程间」共享
            sem_init(&_sem, 0, value);
        }

        Sem(const Sem &) = delete;
        Sem &operator=(const Sem &) = delete;

        // P（荷兰语 Proberen，尝试）：申请一个资源，资源不足则阻塞等待
        void P()
        {
            sem_wait(&_sem);
        }

        // V（荷兰语 Verhogen，增加）：释放一个资源，并唤醒一个等待者
        void V()
        {
            sem_post(&_sem);
        }

        ~Sem()
        {
            sem_destroy(&_sem);
        }

    private:
        sem_t _sem;
    };
}

1.4.2 单生产单消费：完整 RingQueue 代码

光说 P/V 顺序还不够，把环形队列真正写出来才是能用的。Sem 我们已经封装好了（见 1.4.1），下面直接基于它写一个 RingQueue.hpp：

#pragma once
#include <cstddef>
#include <vector>
#include <pthread.h>
#include "Sem.hpp"

using SemModule::Sem;

template <class T>
class RingQueue
{
public:
    explicit RingQueue(size_t cap)
        : _ring_queue(cap),
          _cap(cap),
          _room_sem(cap),   // 生产者关心：还有多少空格
          _data_sem(0),     // 消费者关心：还有多少数据
          _producer_step(0),
          _consumer_step(0)
    {
        pthread_mutex_init(&_producer_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }

    void Enqueue(const T &in)
    {
        // 先预订一个空位置（没有空格就在这里阻塞）
        _room_sem.P();

        // 多生产者时，保护生产下标
        pthread_mutex_lock(&_producer_mutex);
        _ring_queue[_producer_step] = in;
        _producer_step = (_producer_step + 1) % _cap;
        pthread_mutex_unlock(&_producer_mutex);

        // 多了一份可消费数据
        _data_sem.V();
    }

    T Pop()
    {
        // 先预订一份有效数据（没有数据就在这里阻塞）
        _data_sem.P();

        // 多消费者时，保护消费下标
        pthread_mutex_lock(&_consumer_mutex);
        T out = _ring_queue[_consumer_step];
        _consumer_step = (_consumer_step + 1) % _cap;
        pthread_mutex_unlock(&_consumer_mutex);

        // 消费后归还一个空位置
        _room_sem.V();
        return out;
    }

    ~RingQueue()
    {
        pthread_mutex_destroy(&_producer_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }

private:
    std::vector<T> _ring_queue;
    size_t _cap;

    size_t _producer_step;
    size_t _consumer_step;

    Sem _room_sem;  // 生产者关心：还有多少空格
    Sem _data_sem;  // 消费者关心：还有多少数据

    pthread_mutex_t _producer_mutex;
    pthread_mutex_t _consumer_mutex;
};

为什么这里会有两把 mutex？ 因为这是多生产、多消费模型：多个生产线程会同时修改 _producer_step，多个消费线程会同时修改 _consumer_step。

• _producer_mutex：保护"生产者之间"对 _producer_step 的竞争；
• _consumer_mutex：保护"消费者之间"对 _consumer_step 的竞争；
• _room_sem / _data_sem：负责"生产和消费之间"的资源数量与同步（前面游戏规则里说的空/满）。

单生产者 + 单消费者时，这两把索引锁其实可以不要；一旦升级成多生产多消费，就必须处理同类角色之间的竞争。这和阻塞队列里"生产者之间、_psleep_num 也要保护"是一个道理。

这里的顺序是先执行信号量 P 操作，再申请同类线程使用的索引锁。这样做是有意的：如果先拿互斥锁，再因为资源为 0 阻塞在 sem_wait，同类线程可能一直拿不到锁，反而妨碍队列继续推进。

1.4.3 单生产单消费与多生产多消费示例

RingQueue 本身不用改，调整线程数量即可。先看单生产、单消费：

#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include "RingQueue.hpp"

void *Producer(void *args)
{
    auto *rq = static_cast<RingQueue<int> *>(args);
    int value = 0;

    while (true)
    {
        rq->Enqueue(value);
        std::cout << "produce: " << value << std::endl;
        ++value;
        sleep(1);
    }
    return nullptr;
}

void *Consumer(void *args)
{
    auto *rq = static_cast<RingQueue<int> *>(args);

    while (true)
    {
        int value = rq->Pop();
        std::cout << "consume: " << value << std::endl;
        sleep(2);
    }
    return nullptr;
}

int main()
{
    RingQueue<int> rq(5);

    pthread_t p, c;
    pthread_create(&p, nullptr, Producer, &rq);
    pthread_create(&c, nullptr, Consumer, &rq);

    pthread_join(p, nullptr);
    pthread_join(c, nullptr);
    return 0;
}

改成多个生产者和多个消费者时，只需要创建更多线程：

int main()
{
    RingQueue<int> rq(10);

    pthread_t producers[3];
    pthread_t consumers[3];

    for (auto &tid : producers)
        pthread_create(&tid, nullptr, Producer, &rq);

    for (auto &tid : consumers)
        pthread_create(&tid, nullptr, Consumer, &rq);

    for (auto &tid : producers)
        pthread_join(tid, nullptr);

    for (auto &tid : consumers)
        pthread_join(tid, nullptr);

    return 0;
}

信号量的本质是资源的预订机制。基于互斥锁实现阻塞队列时，Enqueue 需要先判断队列是否已满；而在环形队列中，生产者先对空格信号量执行 P 操作。申请成功，说明一定存在空位；申请失败，线程直接等待，不需要再单独编写 IsFull() 判断。

换句话说，信号量把“资源是否存在、是否就绪”的判断提前到了真正访问临界资源之前，并且整个申请过程是原子的。资源可以拆成多份时，信号量更自然；需要保护一个整体对象或一组操作时，互斥锁更直接。

当计数信号量初值为 1 时，它的 P/V 会表现出类似二元信号量 / 互斥门闩的语义（同一时刻只允许一个角色通过），常被用来当互斥锁用；但这不意味着整个 RingQueue 会自动「退化」成 BlockQueue——RingQueue 的环形下标管理、空 / 满两个信号量等结构都还在，只是把其中一个信号量当成二元的来用而已。所以更准确的说法是：可以用二元信号量实现互斥，而不是说环形队列退化成了阻塞队列。

总结

本文从资源计数出发，说明了 P/V 操作的含义，再用环形队列把空格资源和数据资源对应起来。生产者先申请空格、完成写入后发布数据；消费者先申请数据、完成读取后归还空格。

单生产、单消费时，两个信号量已经能够安排生产和消费的先后。扩展到多生产、多消费后，还要分别保护生产下标和消费下标，避免同类线程访问同一个位置。

下一篇进入实战综合篇：日志系统 + 线程池 + 单例模式 + 线程安全与死锁。

---

同系列文章

• 【Linux】线程互斥：从抢票问题到互斥锁与 RAII

参考资料

• man7 - sem_init
• man7 - sem_wait
• man7 - sem_post
• man7 - sem_destroy
• man7 - sem_overview
• cppreference - std::counting_semaphore
• cppreference - std::mutex

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2402_87731470/article/details/164758560