信号量——Linux并发之魂

引言

今天，我们将继续探索Linux线程的基本知识。在上一期博客中，我们详细介绍了条件变量的概念和用法。今天，我们将利用条件变量引出另一个重要话题——信号量的学习。

1.复习条件变量

在上一期博客中，我们没有实际使用条件变量，所以在这里，我们将通过一份代码来复习条件变量的使用。接下来，我们将实现基于BlockingQueue的生产者消费者模型。

1.1 什么是基于BlockingQueue的生产者消费者模型

在多线程编程中，阻塞队列（Blocking Queue）是一种常用于实现生产者和消费者模型的数据结构。它与普通队列的区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入元素；当队列满时，向队列中存放元素的操作也会被阻塞，直到有元素被从队列中取出（这些操作都是基于不同的线程来说的，线程在对阻塞队列进行操作时会被阻塞）。如下图所示：

1.2 分析该模型

我想实现多个生产线程和多个消费线程的模型。我们来分析一下。

首先，生产任务的过程和消费任务的过程必须是互斥关系，不可以同时访问该队列（此时，这个队列是共享资源）。当队列满时，生产线程就不能再生产任务，必须在特定的条件变量下等待；同理，当队列为空时，消费线程就不能再消费任务，也必须在特定的条件变量下等待。所以，类应这样设计：

templateclass BlockQueue{public:    BlockQueue(const int &maxcap=gmaxcap):_maxcap(maxcap)    {        pthread_mutex_init(&_mutex,nullptr);        pthread_cond_init(&_pcond,nullptr);        pthread_cond_init(&_ccond,nullptr);    }    void push(const T&in)//输入型参数，const &    {        pthread_mutex_lock(&_mutex);        while(is_full())        {            pthread_cond_wait(&_pcond,&_mutex);        }        _q.push(in);        pthread_cond_signal(&_ccond);        pthread_mutex_unlock(&_mutex);    }    void pop(T*out)    {        pthread_mutex_lock(&_mutex);        while(is_empty())        {            pthread_cond_wait(&_ccond,&_mutex);        }        *out=_q.front();        _q.pop();        pthread_cond_signal(&_pcond);        pthread_mutex_unlock(&_mutex);    }    ~BlockQueue()    {        pthread_mutex_destroy(&_mutex);        pthread_cond_destroy(&_ccond);        pthread_cond_destroy(&_pcond);    }private:    bool is_empty()    {        return _q.empty();    }    bool is_full()    {        return _q.size()==_maxcap;    }private:    std::queue _q;    int _maxcap; //队列中元素的上限    pthread_mutex_t _mutex;    pthread_cond_t _pcond; //生产者对应的条件变量    pthread_cond_t _ccond;};

由于我们不知道存储的数据类型，所以这里我们选择使用泛型编程的方式。接下来就是要生产任务。为了可以观察到整个生产和消费任务的过程，我们可以生成两个随机数，然后进行运算。代码如下：

class CalTask{    using func_t = function;public:    CalTask() {}    CalTask(int x, int y, char op, func_t func)         :_x(x),_y(y),_op(op),_callback(func)    {}    string operator()()    {        int result=_callback(_x,_y,_op);        char buffer[1024];        snprintf(buffer,sizeof buffer,"%d %c %d=%d",_x,_op,_y,result);        return buffer;    }    string toTaskstring()    {        char buffer[1024];        snprintf(buffer,sizeof buffer,"%d %c %d=?",_x,_op,_y);        return buffer;    }private:    int _x;    int _y;    char _op;    func_t _callback;};const char* oper="+-*/%";int mymath(int x,int y,char op){    int result=0;    switch(op)    {        case '+':            result=x+y;            break;        case '-':            result=x-y;            break;        case '*':            result=x*y;            break;        case '/':            if(y==0)            {                cerr

接下来，我们来写整体的代码。

1.3 完整代码

我们要创建三个文件：BlockQueue.hpp、Task.hpp、Main.cc。各文件内容如下所示：

BlockQueue.hpp

#pragma once#include#include#include#include#include#includeusing namespace std;const int gmaxcap=100;templateclass BlockQueue{public:    BlockQueue(const int &maxcap=gmaxcap):_maxcap(maxcap)    {        pthread_mutex_init(&_mutex,nullptr);        pthread_cond_init(&_pcond,nullptr);        pthread_cond_init(&_ccond,nullptr);    }    void push(const T&in)//输入型参数，const &    {        pthread_mutex_lock(&_mutex);        while(is_full())        {            pthread_cond_wait(&_pcond,&_mutex);        }        _q.push(in);        pthread_cond_signal(&_ccond);        pthread_mutex_unlock(&_mutex);    }    void pop(T*out)    {        pthread_mutex_lock(&_mutex);        while(is_empty())        {            pthread_cond_wait(&_ccond,&_mutex);        }        *out=_q.front();        _q.pop();        pthread_cond_signal(&_pcond);        pthread_mutex_unlock(&_mutex);    }    ~BlockQueue()    {        pthread_mutex_destroy(&_mutex);        pthread_cond_destroy(&_ccond);        pthread_cond_destroy(&_pcond);    }private:    bool is_empty()    {        return _q.empty();    }    bool is_full()    {        return _q.size()==_maxcap;    }private:    std::queue _q;    int _maxcap; //队列中元素的上限    pthread_mutex_t _mutex;    pthread_cond_t _pcond; //生产者对应的条件变量    pthread_cond_t _ccond;};

Task.hpp

#pragma once#include #include #include #include#include using namespace std;class CalTask{    using func_t = function;public:    CalTask() {}    CalTask(int x, int y, char op, func_t func)         :_x(x),_y(y),_op(op),_callback(func)    {}    string operator()()    {        int result=_callback(_x,_y,_op);        char buffer[1024];        snprintf(buffer,sizeof buffer,"%d %c %d=%d",_x,_op,_y,result);        return buffer;    }    string toTaskstring()    {        char buffer[1024];        snprintf(buffer,sizeof buffer,"%d %c %d=?",_x,_op,_y);        return buffer;    }private:    int _x;    int _y;    char _op;    func_t _callback;};const char* oper="+-*/%";int mymath(int x,int y,char op){    int result=0;    switch(op)    {        case '+':            result=x+y;            break;        case '-':            result=x-y;            break;        case '*':            result=x*y;            break;        case '/':            if(y==0)            {                cerr

Main.cc

#include "BlockQueue.hpp"#include "Task.hpp"#include#include#include#includeusing namespace std;void *productor(void *bqs_){    BlockQueue *bqs=static_cast(bqs_);    while(true)    {        int x=rand()%10+1;        int y=rand()%5+1;        int opercode=rand()%(sizeof(oper));        CalTask T(x,y,oper[opercode],mymath);        bqs->push(T);        cout

在代码中，有几个点需要注意一下：第一点：

pthread_cond_wait的第二个参数一定是我们正在使用的互斥锁，这个函数在被运行时，会以原子性的方式将锁释放，然后将自己挂起，等待被条件变量唤醒。该函数在被唤醒时，会自动重新获取持有的锁，然后继续向下执行。假设多个生产者线程一起被唤醒，然后先后持有锁，接着继续生产任务，当队列剩余的空间小于这些生产者生产的任务时，就会出现问题，所以让所有被唤醒的线程先通过while循环，如果有剩余的空间，再进行任务的生产活动。

大家可以在自己试着敲一下，有问题可以在评论区和我交流。接下来，我们来查找一下这些代码有哪些“不足的地方”。

2.代码中的“不足”

一个线程在操作临界资源时，临界资源必须是满足条件的，然后线程才能对临界资源进行操作。比如：在如上代码中，生产者线程只有在队列（临界资源）有剩余空间的条件下，才能进行下一步操作。但是，临界资源是否满足生产和消费的条件，我们不能事前得知，只能在进入临界资源后，再进行进一步的检测。所以，一般访问临界资源的过程为：先加锁，再检测，如果条件满足，就进行下一步的操作；反之，就将该线程挂起，释放锁，然后挂起等待，等到条件满足时，重新获得锁，接着进行下一步操作。因为不可能事先得知是否满足条件，所以我们只能先加锁，进入临界资源内部进行检测。只要我们申请了信号量，就默认对这部分资源的整体使用，但通常情况下，我们使用的仅仅是临界资源的一小部分。实际情况中，有没有可能不同的线程访问临界资源不同部分的情况，有可能。所以，前辈大佬们给出了一种解决方案——信号量。

3.信号量

3.1 什么是信号量

笔魂AI

笔魂AI绘画-在线AI绘画、AI画图、AI设计工具软件

257 查看详情

信号量的本质是一把计数器，一把衡量临界资源多少的计数器。只要拥有信号量，就在未来一定能够拥有临界资源的一部分。

比如：我想去看电影，首先我要买票。我一旦买到票，无论我去不去看电影，都会有一个位置属于我。买票的过程==申请信号量的过程。所以，在访问临界资源之前，我们可以申请信号量。通过申请信号量，我们就可以获知临界资源的使用情况。①只要申请成功，就一定有我可以访问的资源。②只要申请失败，说明条件不就绪，只能等待。如此，就不需要进入临界资源再进行检测了。

3.2 信号量的相关接口

如上这些接口如果调用成功的话，返回0；调用失败的话，返回-1，并且错误原因被设置。我们知道信号量的本质是一把计数器，所以信号量必须可以进行递增和递减的操作。

信号量-1：申请资源，其过程必须是原子性的。简称P操作。信号量+1：归还资源，其过程必须是原子性的。简称V操作。所以，信号量的核心操作：PV原语。

接下来，我们就使用信号量来完成我们的基于环形队列的生产消费模型。

3.3 用信号量来实现基于环形队列的生产消费模型

3.3.1 对环形队列的简单介绍

相信大家在C++学习期间都模拟实现过环形队列。如图：

环形队列的逻辑结构为环形，但其存储结构实际上就是队列，其实就是一个数组，只不过用下标不断的%上队列的长度。大家在模拟实现环形队列时，大家必定遇到的问题是：当rear==front时，究竟是环形队列已满还是环形队列为空呢？其实，这个问题有多种处理方式，今天就不讲了。今天，我们的基于环形队列的生产消费模型必须遵守哪些规则呢？我们来讲一个故事：张三和李四在一个房间里做游戏，这个房间里有一张大圆桌，桌子上有很多的盘子。规定张三往每个盘子里放一个桃子?，然后李四在后边吃桃子?，由于李四还要吃桃子，所以速度一定比张三放的速度慢。

总结一下，我们发现张三和李四必须满足这些规律：

李四不可以超过张三——消费者不可以超过生产者。张三不可以把李四套一个圈——生产者不可以把消费者套一个圈。张三和李四什么时候在一起？①盘子全为空，张三和李四在一起，张三先运行（生产者先运行）。②盘子全为满，张三和李四在一起，李四先运行（消费者先运行）。③其他情况，张三和李四指向不同的位置。

我们将这些规则迁移到环形队列的生产消费模型，就是生产消费模型应该遵守的规则：①消费者不能超过生产者。②生产者不能把消费者套一个圈。③生产者和消费者什么情况下会在一起呢？空的时候和满的时候，对应不同的处理方式。④只要生产者和消费者指向不同的位置，就可以实现生产者和消费者的并发执行。只有在为空和为满时，才会出现同步和互斥问题。

接下来，我们基于这份伪代码来理解一下，看看能否满足我们的规则。

生产者关注的是队列里的剩余空间，在队列为空时剩余空间为10，所以生产者可以顺利申请到信号量。但是由于空间中这部分资源已经被占用，所以无法归还。但是消费者所关注的队列中的数据资源不知不觉中已经多了一份。所以对消费者信号量应进行V操作。

消费者关注的是队列中的数据资源，队列刚开始为空时，数据资源为0，消费者申请失败。等到生产者申请空间成功后，生产了数据。所以消费者可以成功申请到数据资源信号量，然后消费数据。但不知不觉，队列中的剩余空间多了一份，所以应对剩余空间资源的信号量进行V操作。若队列满时，剩余空间信号量为0，生产者申请信号量失败。此时，数据资源信号量为满，消费者可以申请到信号量，从而进行操作。所以必须消费者先运行。若队列空时，数据资源信号量为0，消费者申请信号量失败。此时，剩余空间信号量为满，生产者可以申请到信号量，从而进行操作。所以必须生产者先运行。所以，这伪代码完全符合我们的规则。接下来，我们编写单生产进程和单消费进程的代码。

编写代码我们创建三个源文件：RingQueue.hpp、main.cc、Task.hpp

Ringqueue.hpp:

#pragma once#include #include #include #include #include static const int gcap = 5;templateclass RingQueue{private:    void P(sem_t &sem)    {        int n = sem_wait(&sem);        assert(n == 0); // if        (void)n;    }    void V(sem_t &sem)    {        int n = sem_post(&sem);        assert(n == 0);        (void)n;    }public:    RingQueue(const int &cap = gcap): _queue(cap), _cap(cap)    {        int n = sem_init(&_spaceSem, 0, _cap);        assert(n == 0);        n = sem_init(&_dataSem, 0, 0);        assert(n == 0);        _productorStep = _consumerStep = 0;        pthread_mutex_init(&_pmutex, nullptr);        pthread_mutex_init(&_cmutex, nullptr);    }    // 生产者    void Push(const T &in)    {        // ?: 这个代码 有没有优化的可能        // 你认为：现加锁，后申请信号量，还是现申请信号量，在加锁？        P(_spaceSem); // 申请到了空间信号量，意味着，我一定能进行正常的生产        pthread_mutex_lock(&_pmutex);        _queue[_productorStep++] = in;        _productorStep %= _cap;        pthread_mutex_unlock(&_pmutex);        V(_dataSem);    }    // 消费者    void Pop(T *out)    {        // 你认为：现加锁，后申请信号量，还是现申请信号量，在加锁？        P(_dataSem);        pthread_mutex_lock(&_cmutex);        *out = _queue[_consumerStep++];        _consumerStep %= _cap;        pthread_mutex_unlock(&_cmutex);        V(_spaceSem);    }    ~RingQueue()    {        sem_destroy(&_spaceSem);        sem_destroy(&_dataSem);        pthread_mutex_destroy(&_pmutex);        pthread_mutex_destroy(&_cmutex);    }private:    std::vector _queue;    int _cap;    sem_t _spaceSem; // 生产者 想生产，看中的是什么资源呢? 空间资源    sem_t _dataSem;  // 消费者 想消费，看中的是什么资源呢? 数据资源    int _productorStep;    int _consumerStep;    pthread_mutex_t _pmutex;    pthread_mutex_t _cmutex;};

Task.hpp

#pragma once#include #include #include #include class Task{    using func_t = std::function;    // typedef std::function func_t;public:    Task()    {}    Task(int x, int y, char op, func_t func)    :_x(x), _y(y), _op(op), _callback(func)    {}    std::string operator()()    {        int result = _callback(_x, _y, _op);        char buffer[1024];        snprintf(buffer, sizeof buffer, "%d %c %d = %d", _x, _op, _y, result);        return buffer;    }    std::string toTaskString()    {        char buffer[1024];        snprintf(buffer, sizeof buffer, "%d %c %d = ?", _x, _op, _y);        return buffer;    }private:    int _x;    int _y;    char _op;    func_t _callback;};const std::string oper = "+-*/%";int mymath(int x, int y, char op){    int result = 0;    switch (op)    {    case '+':        result = x + y;        break;    case '-':        result = x - y;        break;    case '*':        result = x * y;        break;    case '/':    {        if (y == 0)        {            std::cerr

main.cc

#include "RingQueue.hpp"#include "Task.hpp"#include #include #include #include #include std::string SelfName(){    char name[128];    snprintf(name, sizeof(name), "thread[0x%x]", pthread_self());    return name;}void *ProductorRoutine(void *rq){    // RingQueue *ringqueue = static_cast(rq);    RingQueue *ringqueue = static_cast(rq);    while(true)    {        // version1        // int data = rand() % 10 + 1;        // ringqueue->Push(data);        // std::cout

大家可以自己敲一敲，试一下。写到这里，这篇博客就结束了，下篇博客我们再见。

以上就是信号量——Linux并发之魂的详细内容，更多请关注创想鸟其它相关文章！