手撕 Linux 信号量：从古老的 PV 原语到现代内核，极致简洁的同步美学

前言： 在前面，我们讲解了线程互斥和线程同步的相关内容，但是既然谈到了线程的互斥与同步，就不得不提信号量了，这也是线程的互斥和同步这部分内容中不可忽视的一部分，那么今天我们就来深入探讨一下信号量，下面我们来一起看看吧。

一.信号量的基本概念

我们要想理解什么是信号量，就要先了解什么是对资源的整体使用和对资源的局部使用，我们来看：

在前面的章节中我们讲过ATM机的例子，现在我们在拿它来举例，ATM机这种小房间就是一个很好的对资源整体使用的例子，这个房间中一次只能进去一个人，别人要想进去就只能等里面的人出来，这个小房间的资源就只供一个人去使用。

而电影院想必我们每个人都去过，电影院就是一个很好的对资源进行局部使用的例子，在这间房间中，不只有我们自己可以看电影，座位上的每个人都可以观看电影，而我们通过座位就将电影院的这整个资源给划分为了一个个的局部资源。

通过上面的例子，我们对于资源的整体使用和局部使用有了一个的简单理解，我们今天的重点不在于对资源的整体使用，而在于对资源的局部使用。

我们来思考一个问题：我们在看电影之前或者说我们要看电影，我们首先要干什么？ 答案很明显，我们要买票，我们买票是为了什么呢？ 为了证明这个座位资源是我的，那么：对于这个座位资源，我们是买了票之后，这个座位就是我的，还是我坐到座位上这个资源才是我的呢？ 答案是我们买了票，这个座位就是我的，不管我今天去不去，这个座位就只能我坐，所以我们买票的本质就是：对资源的预定机制！！！

有了对上面的认识后，我们在来思考一个问题：作为一家电影院的老板，你最担心的是什么呢？

答案其实就两点：

1.多卖票。明明电影院就100个座位，却卖了110张票，那么就必然就有人没有座位，自然就会起冲突，没一个电影院的老板想看到这种现象。 2.卖重复的票。一个座位按理说只能卖一张电影票，但却卖了10张，那么导致的后果就是这10个人争夺这一个座位，出现这种情况可谓是灾难性的。

对于这第二种卖重复的票我们暂且先不提，我们先着重说这第一种情况：我们在买票的时候，都会有一个售票系统，那么售票系统是怎么知道票是否卖完了呢？

答案就是这个售票系统的底层实现中，一定会有一个计数器，这个计数器记录了电影院有多少个座位，有多少个座位就卖多少张票，也就是这个计数器描述了电影院座位资源的多少。

而我们今天要讲的信号量，本质就是一个计数器，一个描述临界资源多少的计数器！！！

我们现在将视角转移到计算机中，我们可以通过信号量这个计数器，就可以保证：

1.资源不会出现多申请的情况，也就是上面的多卖票的情况。 2.所有的进程或线程未来想进入临界区，去访问临界资源，都要先申请信号量，就如同上面我们要先买票一样。

既然是一个计数器，那么不可避免的我们就要对其进++和--的操作，--操作就表示我们申请了一份资源，++操作表示我们归还了一份资源。

而对于信号量的++和--操作，我们对其有更优雅的称呼：--操作我们称之为P操作，++操作我们称之为V操作。

讲到这里可能有人就问了：每个线程都要申请信号量的话，那么前提就是所有的线程都要看到信号量，也就是信号量本身就是共享资源，它保护了临界资源的安全，该怎么保护自己的安全呢？

答案就是我们改变信号量的方式就是通过PV操作，那换句话说，我们保证了PV操作的安全，也就保证了信号量的安全，那么该如何保证PV操作的安全呢？ 我们知道PV操作也就是--和++的操作，这种操作是可以被打断的，所以我们的做法就是让PV操作具有原子性，让--和++的过程不可被打断，这样就保证了PV操作的安全，也就保证了信号量的安全。

二.快速认识一下信号量接口

下面我们就来看看信号量的接口都有哪些：

那么第一个函数就是：sem_init，这个函数我们看名字就知道就是对信号量进行初始化。

而它的返回值很简单，成功就返回0，失败了就返回-1并且设置错误码。

下面我们来介绍它的三个参数：

1.sem_t *sem 这个参数的作用很简单，我们既然想对信号量进行初始化，那得指明你要对那个信号量进行初始化吧，所以该参数的作用就是指明要初始化的对象。 2.int pshared 这个参数就有意思了，这个参数是一个int类型的参数，根据POSIX标准，这个函数的类型分为0和非0这两类。 当该参数为0时，表示该信号量是进程内私有的，它只能被同一个进程内的不同线程使用，简单理解就是当前的信号量是在线程之间使用的。 当该参数为非0时，表示该信号量是进程间共享的，也就是在进程间去使用。 3.unsigned int value 既然信号量是一个计数器，那我们总要给它设置一个起始值吧，所以该参数就是来设置信号量的初始值。

既然对一个信号量进行了初始化，那么我们不用的时候就要将其销毁，所以要用到的函数就是：sem_destroy。

它的返回值和上面的sem_init一样，都是成功了返回0，失败了就返回-1，并设置错误码。

那么下面就是要对信号量进行操作的函数了，第一个就是：sem_wait。

我们只看这个函数并不知道它是什么意思，而第二张图中的decrements就揭示了它的作用，没错，就是：P操作，也就是对信号量进行--操作。

那么与之相对应的就是：sem_post，上面的sem_wait函数是P操作，那么该函数就是就是：V操作，也就是对信号量进行++操作。

三.基于环形队列的生产者消费者模型

下面我们就要通过信号量来实现一个基于环形队列的生产者消费者模型，那么首先我们先来简单介绍一下环形队列。

3.1环形队列的简单认识

上面就是环形队列的逻辑图，刚开始首和尾是在同一个位置的，之后向里面写入数据，为就开始向后移动，但是向队列中填入最后一个数据后，首和尾又会重新指向同一个位置，也就是首尾指针如果指向同一个位置，就表明：为空或者为满，而如果首和尾指向不同的位置，就表明：首和尾访问的一定不是同一个位置。

那首尾相连是如何做到的呢？我们来看：

这里我用数组来模拟环形队列，刚开始首和尾都在数组的起始位置，那么尾要想最后重新回到数组的起始位置，那么就可以这样做：

通过简单的模运算我们就可以让尾重新回到数组的起始位置。

那么我们现在将视角切换到生产者消费者模型，下面我们来思考一些问题：

1.队列为空的时候，先让生产者先运行还是消费者先运行？ 2.队列为满的时候，先让生产者先运行还是消费者先运行？ 3队列不为空&&不为满，又该如何？

下面我们就来对这三个问题进行解析：

第一个问题： 毋庸置疑，队列为空时一定是先让生产者先运行，没有数据消费者怎么消费，并且生产者正在放入数据的过程是不能被打扰的，换句话说就是生产者要互斥的向队列中放入数据，因为此时的生产者和消费者访问的是同一个位置，我们要避免并发问题。 第二个问题： 和第一题正好相反，队列为满时一定是先让消费者先运行，因为生产者接着生产的话就会将数据覆盖掉，这肯定是不行的，并且和上面一样，消费者正在取出数据的过程是不能被打扰的，所以消费者也是互斥的从队列中取出数据。 在上面的两个问题的回答中我们说明了生产者和消费者之间的互斥关系，但是它们之间并不只有互斥的关系，就像上面我们说的：消费者不能超过生产者，生产者不能套圈消费者，它们之间是有一定的顺序性的，所以这里面还体现了生产者和消费者之间的同步关系。 第三个问题： 上面所出现的生产者和消费者之间的互斥与同步与互斥关系是因为它们访问的是同一个位置，而我们现在的情况是队列不为空也不为满，那么生产者和消费者访问的就不是同一个位置，既然不是同一个位置，那我们不就可以并发执行了吗？

那么现在有了上面的逻辑后，我们该如何实现它呢？

要想实现上面的逻辑我们要先弄清楚两点：

1.生产者最关心什么？

2.消费者最关心什么？

第一个问题： 很明显，生产者最关心的应是剩余的空间资源，剩余的空间资源没有了生产者就不能再生产了。 第二个问题： 消费者最关心的应是当前的数据资源，如果当前已经没有数据资源了，那么消费者也就不能再消费了。

所以下面我们简单写一个伪代码，整理一下整体的逻辑：

既然生产者和消费者关心的东西并不相同，所以我们实现时就要定义两个信号量，分别记录剩余的空间资源和当前的数据资源。

并且对于生产者而言，当生产了数据之后，就相当于队列中的数据资源就变多了，所以它的V操作要对消费者的信号量使用，反之消费者也是一样，消费者消费完数据后，相当于剩余的空间资源就变多了，所以它的V操作就要对生产者的信号量使用。

3.2代码实现

下面我们依旧是老规矩，先实现单生产者单消费者的RingQueue环形队列。

3.2.1准备工作

void *consumer(void *args)
{
    RingQueue<int> *rq = (RingQueue<int> *)args;
    while (true)
    {
        int data;
        rq->Pop(&data);
        cout << "消费者消费了一个数据: " << data << endl;
    }
}

void *productor(void *args)
{
    RingQueue<int> *rq = (RingQueue<int> *)args;
    int data = 1;
    while (true)
    {
        rq->Enqueue(data);
        cout << "生产者生产了一个数据: " << data << endl;
        data++;
    }
}

int main()
{
    RingQueue<int> *rq = new RingQueue<int>();
    pthread_t t1, t2;
    pthread_create(&t1, nullptr, consumer, (void *)rq);
    pthread_create(&t2, nullptr, productor, (void *)rq);

    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    
    delete rq;
    return 0;
}

class Sem
{
public:
    Sem(int num):_initnum(num)
    {
        sem_init(&_sem,0,_initnum);
    }

    void P()
    {
        int n = sem_wait(&_sem);
        (void)n;
    }

    void V()
    {
        int n = sem_post(&_sem);
        (void)n;
    }

    ~Sem()
    {
        sem_destroy(&_sem);
    }
private:
    sem_t _sem;
    int _initnum;
};

对于这三个准备工作我们从上往下依次来讲解：

第一个是我们的测试代码，和上一篇我们实现BlockingQueue时写的测试代码几乎一样，里面所用到的各种函数也是我们之前都见过的，所以这里就不再赘述了。 而对于第二个的信号量Sem，这里我没有向上一篇一样直接用库中的函数，这里我们换一种玩法，直接一步到位，自己封装一个信号量的类，这样使用起来也更加的方便和美观。 虽然是我们自己封装实现的，但是我们可以看到实现起来很简单，直接在我们封装的方法内部直接调用库函数即可，没什么难度，就不过多解释了。 第三个是我们封装实现的RingQueue环形队列，上面我们是按照数组来讲的，所以这里我们也用数组来实现，除此之外还要定下队列的容量，以及两个信号量，这是我们上面讲过的。

有了上面的准备工作后，下面我们就可以来完善环形队列中剩下的内容了。

3.2.2完善环形队列的代码

而当我们想生产数据的时候发现了一个问题，我们该向数组的什么位置放入数据呢？

所以我们要用两个变量来记录生产者和消费者生产和消费的位置，刚开始我们将其初始化为0，表示从数组的起始位置开始。

有了这点的补充后，我们就可以完成上面的工作了：

这样我们直接就可以完成Enqueue函数和Pop函数的实现了，代码很简单，对于信号量的PV操作上面我们也说清楚了，这里就不在赘述了。

至此代码我们就全部实现完了，下面我们用上面写的测试代码跑一下看看效果：

这里我们让消费者先休眠1秒再进行消费，那么我们看到的效果就应该是：生产者直接就将队列给填满了，1秒过后，消费者消费一条数据，生产者生产一条数据，是否真的如此呢？我们来看：

答案如我们所料，前1秒，生产者瞬间就生产了5条数据，1秒过后，消费者开始消费，并且是消费者消费一条数据，生产者生产一条数据。

至此我们就完成了RingQueue环形队列的实现。

3.3解答疑问

对于上面实现RingQueue环形队列的例子相信我们对于它还有不少问题，下面我用几个问题来总结：

1.在上面的代码中，我们并没有使用互斥锁来进行约束，这能保证数据的安全吗？

2.我们并没有在临界区内部判断资源是否就绪，为什么呢？

第一个问题： 这个问题的答案其实在我们已经说明了答案，队列为空时，消费者不能消费，也就是此时的消费者申请不了信号量，队列为满时，生产者不能生产，生产者就申请不了信号量，它们均会在申请信号量时被阻塞，直到队列不为空&&不为满才能申请信号量。 故我们使用信号量，已经变相地完成了为空和为满的同步与互斥动作！！！ 第二个问题： 我们虽然没有主动去判断资源是否就绪，但是我们对信号量进行P操作，也就是申请信号量时，其实已经对资源是否就绪进行了判断。 申请不了信号量，被阻塞了，不就说明资源没有准备就绪吗？成功申请了信号量，不就说明资源准备就绪了吗？ 我们只是将判断从临界区内部转移到了临界区外部(或者入口处)！！！

不止上面的两个问题，我们再来思考一个问题：如果将信号量的值设为1，为怎么样呢？

那么就说明资源只有一份，同一时间，生产者和消费者只有一个能进入临界区去执行代码，那这不就是锁的功能吗？

所以将信号设置为1，也就是二元信号量，其本质就是锁，所以我们此时就来重新理解一下锁：就是认为自己的资源只有一份，申请锁就相当于信号量的P操作，而释放锁就相当于信号量的V操作。

所以我们就可以认为：锁就是信号量的一种特殊情况！！！

3.4单生产者单消费者 -> 多生产者多消费者？

上面只是单生产者单消费者的模型，如果是多生产者多消费者呢？

上面我们是单生产者单消费者，所以我们只需要维护生产者和消费者之间的互斥与同步就行了，但是此时变为了多生产者多消费者，所以我们还要维护生产者与生产者之间的互斥关系，消费者与消费者之间的互斥关系！！！

那么做法就很明显了，我们要加锁，不然可能会因为并发问题导致数据不一致的问题，那么该加几把锁？怎么加呢？

class Mutex
{
public:
    Mutex()
    {
        pthread_mutex_init(&_lock, nullptr);
    }
    void Lock()
    {
        pthread_mutex_lock(&_lock);
    }
    void Unlock()
    {
        pthread_mutex_unlock(&_lock);
    }
    pthread_mutex_t *Get()
    {
        return &_lock;
    }
    ~Mutex()
    {
        pthread_mutex_destroy(&_lock);
    }
private:
    pthread_mutex_t _lock;
};

这里我们采用的做法依旧是自己封装一个Mutex类，并且创建两把锁，我们曾经只创建一把锁是因为我们只需要维护生产者和消费者之间的互斥关系，现在却要维护生产者和生产者之间的互斥关系，消费者和消费者之间的互斥关系，需要维护的互斥关系变多了，所以锁的数量也要增多。

那么下面我们就用多生产者多消费者的例子来试验一下看看效果：

从上面可以看到，加了锁之后确实可行，上面串行问题我们不必在意，毕竟我们并没有保护显示器文件，出现串行是正常现象。

那么最后我们来思考一个问题：我们上面是把锁放在了信号量外面，如果放在里面呢？这两种放置方式我们该选择哪种呢？

我们先将锁放在信号量里面先看看效果：

从结果中我们可以看到，这样做同样也没有问题，所以我们该如何在这两种方式中进行选择呢？

我们先看第一种，如果是在信号量外面加上了锁，那么我们整体的过程就变为了串行，用电影院来说明的话就是我们不仅买票要排队一个一个买，进场也得一个一个进。 但是如果是第二种，就相当于我们是在手机上买票，不用排队，只要有票，我们就可以直接买，然后进场是一个一个进，和上面一样。

那么通过这样的对比，相信大家很快就能知道最佳实践是哪种方式了，没错，就是在信号量里面进行加锁，这种方式可以增加并发度，在效率上是要比第一种方式高的。

const int capcity = 5;

template <class T>
class RingQueue
{
public:
    RingQueue(int cap = capcity) : _cap(cap), _ringqueue(cap), _space_sem(cap), _data_sem(0),_p_index(0),_c_index(0)
    {
    }

    void Pop(T *out)
    {
        _data_sem.P();

         _c_mutex.Lock();
        //消费数据
        *out = _ringqueue[_c_index++];
        //维持环形特点
        _c_index %= _cap;
        _c_mutex.Unlock();

        _space_sem.V();
    }

    void Enqueue(const T &in)
    {
        _space_sem.P();

        _p_mutex.Lock();
        // 生产数据
        _ringqueue[_p_index++] = in;
        //维持环形特点
        _p_index %= _cap;
        _p_mutex.Unlock();
        
        _data_sem.V();
    }

    ~RingQueue()
    {
    }

private:
    vector<T> _ringqueue;
    int _cap;

    Sem _space_sem;
    Sem _data_sem;

    int _p_index; // 生产者生产数据的位置
    int _c_index; // 消费者消费数据的位置

    Mutex _p_mutex;
    Mutex _c_mutex;
};

最后奉上RingQueue环形队列的完整实现代码。

以上就是手撕 Linux 信号量：从古老的 PV 原语到现代内核，极致简洁的同步美学的全部内容。