【Linux】生产消耗模子实践 --- 基于信号量的环形队列

生产消费者模型里，环形队列 + 双信号量是最经典的写法：一个信号量管「还剩多少空间」，一个管「还有多少数据」。在这之上再加两把锁，就能支持多生产多消费。这篇把从原理到代码整条链路拆开讲。

一、信号量回顾

信号量本质是一个计数器，初始化时设定资源数目，获取信号量才能操作资源，结束操作释放信号量。它和管道不同：管道以传数据为主要目的，信号量是用来保护共享资源的（信号量自己也是临界资源），让资源在同一时刻只被一个进程独享；当资源只有一个时，它就等价于一把互斥锁。

信号量只有两种操作，合称 PV 操作：

• P 操作（获取/等待）：本质是计数器减 1。sv 大于 0 就减 1；等于 0 就把该进程挂起。对应接口：sem_wait()（阻塞）、sem_trywait()（非阻塞一次）、sem_timedwait()（限时等待，超时不再阻塞）。
• V 操作（释放）：本质是计数器加 1。如果有进程因等这个信号量被挂起，就唤醒它；没有就单纯加 1。对应接口 sem_post()。

int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem);
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);
int sem_post(sem_t *sem);

PV 都是原子的，不用操心线程安全。初始化 sem_init(sem, pshared, value) 里 pshared 传 0 表示线程间共享、非 0 表示进程间共享；sem_destroy() 负责销毁。

二、环形队列的骨架

用 vector 模拟环形队列，成员变量有这么几个：

• 底层线性容器 _rq，用下标取模模拟环。
• 最大容量 _max_cap。
• 消费者下标 _c_step、生产者下标 _p_step。
• 两个信号量：sem_t _data_sem 表示当前有多少数据，sem_t _space_sem 表示还剩多少空间。

构造函数里：容量给值，两个下标都置 0，_space_sem 初始化为最大容量，_data_sem 初始化为 0；再顺手把两把锁 pthread_mutex_init 起来。

三、Push 与 Pop：一进一出怎么配信号量

3.1 Push（生产）

1. 先 P 获取空间信号量 sem_wait(&_space_sem)，拿不到就说明满了，阻塞等着（获取信号量是原子的）。
2. 进临界区加锁，把数据写到 _rq[_p_step]，下标后移并对容量取模，防止越界。
3. 解锁，再 V 释放数据信号量 sem_post(&_data_sem)，通知消费方有货了。

3.2 Pop（消费）

1. 先 P 获取数据信号量 sem_wait(&_data_sem)，没数据就阻塞。
2. 进临界区加锁，取出 _rq[_c_step]，下标后移对容量取模。
3. 解锁，V 释放空间信号量 sem_post(&_space_sem)，告诉生产方可以继续产了。

规律就一句：往里放就消耗空间、生产数据；往外取就消耗数据、释放空间。两个信号量的方向永远相反，配好了自然不会越界。

四、为什么多生产多消费还要加锁

单生产单消费时，两个信号量就能保证不越界、不重叠。但多线程下问题来了：抢到信号量之后，中间那段处理是临界区，随时可能发生线程切换；如果这时另一个线程也拿着信号量往同一个下标写，数据就被覆盖。

解决办法是加两把锁，一把管生产者、一把管消费者：

• 获取信号量之后再加锁。因为获取信号量是原子的，先抢信号量能让多个线程在队列里有序排队等待。
• 反过来先加锁会怎样？只有一个线程能进到「获取信号量」这一步，其余全被挡在锁外，并发度直接塌掉。打个比方：电影院应该先买票再排队，而不是先排死队再想买不买票。

这就是「锁的粒度」和「锁的顺序」两个问题：信号量管「能不能做」，锁管「同一时刻谁做」。

五、完整代码

#pragma once

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

const int default_cap = 5;

template <class T>
class RingQueue
{
public:
    RingQueue(int max_cap = default_cap)
        : _rq(max_cap), _max_cap(max_cap), _p_step(0), _c_step(0)
    {
        sem_init(&_space_sem, 0, _max_cap);
        sem_init(&_data_sem, 0, 0);
        pthread_mutex_init(&_c_mtx, nullptr);
        pthread_mutex_init(&_p_mtx, nullptr);
    }

    void P(sem_t &sp)  { sem_wait(&sp); }
    void V(sem_t &sp)  { sem_post(&sp); }

    void Push(const T &t)
    {
        P(_space_sem);                        // 申请空间++
        pthread_mutex_lock(&_p_mtx);          // 临界区上锁
        _rq[_p_step] = t;
        _p_step++;
        _p_step %= _max_cap;
        pthread_mutex_unlock(&_p_mtx);
        V(_data_sem);                         // 释放数据++
    }

    void Pop(T *t)
    {
        P(_data_sem);                         // 申请数据++
        pthread_mutex_lock(&_c_mtx);
        *t = _rq[_c_step];
        _c_step++;
        _c_step %= _max_cap;
        pthread_mutex_unlock(&_c_mtx);
        V(_space_sem);                        // 释放空间++
    }

    ~RingQueue()
    {
        sem_destroy(&_space_sem);
        sem_destroy(&_data_sem);
        pthread_mutex_destroy(&_c_mtx);
        pthread_mutex_destroy(&_p_mtx);
    }

private:
    std::vector<T> _rq;
    int _max_cap;
    int _p_step;
    int _c_step;
    sem_t _space_sem;
    sem_t _data_sem;
    pthread_mutex_t _p_mtx;
    pthread_mutex_t _c_mtx;
};

六、测试运行

测试时用 Task 类模拟任务：它带两个整型参数 _x 和 _y，重载了括号运算符，对象可以直接当函数调用执行加法并返回结果，同时提供调试信息输出。

线程函数就两个：Consumer 不断从环形队列 Pop 出 Task 并执行、打印结果；Productor 持续生成 Task 塞进队列、打印生产信息。main 里建一个容量为 5 的 RingQueue，pthread_create 起两个线程，pthread_join 等它们收尾，以此验证多线程下的线程安全性和结果正确性。

七、为什么这里不用条件变量

标准块级队列常用条件变量 + while 判断，但环形队列这里「不管三七二十一直接 P」，因为信号量本身就是判断条件：它描述内部资源的多少，且是原子的，通过预订机制保证资源被合理使用。当资源数目为 1 时，信号量行为和锁完全等价——所以能用信号量表达的判断，就别再套一层 while 判断了。

八、小结

环形队列生产消费模型的核心组合是：两个信号量控数量（空间 / 数据）+ 两把锁控位置（生产 / 消费）。顺序永远是「先 P 信号量，再拿锁改下标，最后 V 对面的信号量」。把这三条记牢，多生产多消费也不会出错。

qidao123.com企服评测-应用市场 编辑提示