? 本篇文章重點對信號量的概念，信號量的申請、初始化、釋放、銷毀等操作進(jìn)行講解。同時舉例把信號量應(yīng)用到生產(chǎn)者消費者模型來理解。希望本篇文章會對你有所幫助。

目錄

一、信號量概念

1、1 什么是信號量

1、2 為什么要有信號量

1、3 信號量的PV操作

二、信號量的相關(guān)接口

2、1 sem_t

2、2 sem_init

2、3 sem_wait

2、4?sem_post

2、5 sem_destory

三、基于信號量的生產(chǎn)者消費者模型

3、1 信號量控制環(huán)形隊列

3、1、1 空間資源和數(shù)據(jù)資源

3、1、2 保護原理（二元信號量）

3、2 demo代碼

3、2、1 單生產(chǎn)與單消費

?3、2、2 多生產(chǎn)多消費

---

???♂??作者：@Ggggggtm????♂?

???專欄：Linux從入門到精通? ??

???標(biāo)題：信號量??

????寄語：與其忙著訴苦，不如低頭趕路，奮路前行，終將遇到一番好風(fēng)景???

一、信號量概念

1、1 什么是信號量

? 我們之前學(xué)了互斥鎖和體哦阿健變量可以實現(xiàn)線程的互斥與同步。那么還有其他方法嗎？信號量也可以做到！

??信號量（Semaphore）是操作系統(tǒng)中一種用于實現(xiàn)線程間同步與互斥的機制。它本質(zhì)就是一個計數(shù)器，用于控制多個線程對共享資源的訪問。信號量可以被視為一個簡單的整數(shù)變量，并且可以進(jìn)行原子操作，包括等待（wait）和釋放（signal）。

1、2 為什么要有信號量

? 信號量（Semaphore）是一種多線程同步的機制，用于解決并發(fā)環(huán)境中的資源競爭問題。在并發(fā)編程中，多個線程可能同時訪問共享資源，如果不對資源進(jìn)行合理的管理，就會導(dǎo)致數(shù)據(jù)不一致或錯誤的結(jié)果。

? 我們在學(xué)習(xí)互斥鎖時，一個線程在操作臨界資源的時候，必須臨界資源是滿足條件的!可是公共資源是否滿足生產(chǎn)或者消費條件，我們無法直接得知。因為你要檢測，本質(zhì)也是在訪問臨界資源。所以只能先加鎖，再檢測，再操作，再解鎖。只要我們對資源進(jìn)行整體加鎖，就默認(rèn)了我們對這個資源整體使用。但是，有時候會是一份臨界資源同時訪問不同的區(qū)域。這時互斥鎖并不能很好的滿足對臨界資源的充分利用。在這種情況下就可以引入信號量來很好的解決。具體如下圖：

? 現(xiàn)在我們有一個共享資源，不當(dāng)做一個整體，而讓不同的執(zhí)行流訪問不同的區(qū)域的話，那么不就可以繼續(xù)并發(fā)了。

1、3 信號量的PV操作

? 這里會有一個疑問，我們怎么知道臨界資源內(nèi)部一共有少個區(qū)域資源呢？我們又怎么知道內(nèi)部一定還有資源呢？實際上，一般都是外部就會提供有多少資源。同時，信號量一定會保證內(nèi)部是否還有資源。

? 信號量本質(zhì)是一個計數(shù)器。一個線程在申請信號量，本質(zhì)就是在對信號量的 -- 操作（對剩余資源數(shù)量的減減操作）。只要擁有信號量，就在未來一定能夠擁有臨界資源的一部分。申請信號量的本質(zhì):對臨界資源中特定小塊資源的預(yù)訂機制。信號量因此保證了只要你申請成功，就代表一定還有資源。如果申請失敗，就會進(jìn)入等待。這不就是我們在訪問真正的臨界資源之前，我們其實就可以提前知道臨界資源的使用情況?。?！就不用再進(jìn)行復(fù)雜的加鎖、判斷、解鎖等操作了。

? 此時發(fā)現(xiàn)，線程要訪問臨界資源中的某一區(qū)域，就得先申請信號量。所有人必須的先看到統(tǒng)一信號量。信號量本身必須是公共資源。

? 對信號量的操作就是申請和釋放操作。信號量本質(zhì)就是一個計數(shù)器，也就是在對信號量進(jìn)行++和-- 操作。申請資源，可以看成對sem--，同時必須保證操作的原子性，我們也稱之為 P 操作。釋放資源，可以看成sem++，也必須保證操作的原子性，稱之為 V?操作 。信號量核心操作:PV操作。

二、信號量的相關(guān)接口

2、1 sem_t

? sem_t 是在 POSIX 系統(tǒng)中用來實現(xiàn)信號量機制的類型。它是一個不透明的數(shù)據(jù)結(jié)構(gòu)，用于控制多個進(jìn)程或線程對共享資源的訪問。

??sem_t 提供了三個主要的函數(shù)接口：

1. sem_init：用于初始化一個信號量。該函數(shù)接受三個參數(shù)，分別是指向 sem_t 對象的指針、信號量的共享標(biāo)志和初始值。共享標(biāo)志指定信號量的共享方式，根據(jù)具體需求可以選擇在進(jìn)程間共享（設(shè)置為0）或者在同一進(jìn)程內(nèi)的線程間共享（設(shè)置為非0）。初始值表示信號量的初始計數(shù)值。

2. sem_wait：該函數(shù)使調(diào)用線程等待信號量。如果信號量的計數(shù)值大于0，則將計數(shù)值減一，并立即返回。如果計數(shù)值為0，則線程將阻塞，直到信號量的計數(shù)值大于0。

3. sem_post：該函數(shù)用于釋放信號量。它將信號量的計數(shù)值加一，并喚醒因等待該信號量而阻塞的線程。

4. sem_destroy：該函數(shù)是用于銷毀一個已經(jīng)初始化的信號量的函數(shù)，在使用完信號量后，通過調(diào)用該函數(shù)可以釋放相關(guān)資源。

? 下面我們來看這幾個函數(shù)的詳細(xì)解釋。

2、2 sem_init

? sem_init函數(shù)是用于初始化一個信號量的函數(shù)，它在程序中創(chuàng)建一個新的信號量，并為其分配必要的資源。函數(shù)原型如下：

int sem_init(sem_t *sem, int pshared, unsigned int value);

參數(shù)： sem_init函數(shù)有三個參數(shù)：

• sem：一個指向sem_t類型的指針，用于存儲初始化后的信號量對象。
• pshared：表示信號量的共享方式。
  • 如果pshared的值為0，表示信號量只能在調(diào)用它的進(jìn)程內(nèi)的線程之間共享。
  • 如果pshared的值為非零，表示信號量可以在多個進(jìn)程之間共享。
• value：表示信號量的初始值（初始臨界資源內(nèi)部有多少個小塊資源）。

返回值： sem_init函數(shù)的返回值是一個整數(shù)，用于表示函數(shù)調(diào)用是否成功。

• 如果返回值為0，表示函數(shù)調(diào)用成功。
• 如果返回值為-1，表示函數(shù)調(diào)用失敗，此時可以通過查看全局變量errno獲取錯誤碼。

??以下是一個示例代碼，演示了如何使用sem_init函數(shù)來初始化一個信號量：

#include <stdio.h>
#include <semaphore.h>

int main() {
    sem_t mySem;
    
    // 初始化一個非共享的信號量，初始值為1
    int ret = sem_init(&mySem, 0, 1);
    if (ret == -1) {
        perror("Failed to initialize semaphore");
        return 1;
    }

    // 進(jìn)行其他操作...

    // 銷毀信號量
    sem_destroy(&mySem);

    return 0;
}

2、3 sem_wait

? sem_wait函數(shù)用于對信號量進(jìn)行等待操作，同時會將信號量減1。以實現(xiàn)對臨界資源的互斥訪問。函數(shù)原型如下：

int sem_wait(sem_t *sem);

參數(shù)： sem_wait函數(shù)只有一個參數(shù)：

• sem：一個指向已經(jīng)初始化的信號量對象。

返回值： sem_wait函數(shù)的返回值是一個整數(shù)，用于表示函數(shù)調(diào)用是否成功。

• 如果返回值為0，表示函數(shù)調(diào)用成功，信號量的值被成功減一。
• 如果返回值為-1，表示函數(shù)調(diào)用失敗，此時可以通過查看全局變量errno獲取錯誤碼。常見的錯誤碼包括EINTR（被信號中斷）和EDEADLK（死鎖）等。

2、4?sem_post

? sem_post函數(shù)用于對信號量進(jìn)行發(fā)布操作，以增加信號量的值（對信號量加1）。它通常與sem_wait函數(shù)一起使用，用于在對共享資源的訪問結(jié)束后釋放信號量，以便其他線程可以獲取到該資源。函數(shù)原型如下：

int sem_post(sem_t *sem);

參數(shù)： sem_post函數(shù)只有一個參數(shù)：

• sem：一個指向已經(jīng)初始化的信號量對象。

返回值： sem_post函數(shù)的返回值是一個整數(shù)，用于表示函數(shù)調(diào)用是否成功。

• 如果返回值為0，表示函數(shù)調(diào)用成功，信號量的值被成功增加。
• 如果返回值為-1，表示函數(shù)調(diào)用失敗，此時可以通過查看全局變量errno獲取錯誤碼。常見的錯誤碼包括EINVAL（信號量未初始化）和EOVERFLOW（信號量值達(dá)到上限）等。

? 需要注意的是，sem_post函數(shù)并不處理過度發(fā)布的情況，即如果信號量的值已經(jīng)達(dá)到了其上限，再調(diào)用sem_post函數(shù)也無法將其繼續(xù)增加。因此，在使用信號量時，必須正確地控制信號量的值以避免出現(xiàn)競態(tài)條件或死鎖等問題。

2、5 sem_destory

? sem_destory函數(shù)用于銷毀一個已經(jīng)初始化的信號量對象，并釋放相關(guān)的資源。當(dāng)不再需要使用信號量時，應(yīng)該調(diào)用sem_destroy函數(shù)進(jìn)行清理操作。函數(shù)原型如下：

int sem_destroy(sem_t *sem);

參數(shù)： sem_destroy函數(shù)只有一個參數(shù)：

• sem：一個指向已經(jīng)初始化的信號量對象。

返回值： sem_destroy函數(shù)的返回值是一個整數(shù)，用于表示函數(shù)調(diào)用是否成功。

• 如果返回值為0，表示函數(shù)調(diào)用成功，信號量對象被成功銷毀并釋放了相關(guān)的資源。
• 如果返回值為-1，表示函數(shù)調(diào)用失敗，此時可以通過查看全局變量errno獲取錯誤碼。常見的錯誤碼包括EINVAL（信號量未初始化）和EBUSY（仍有線程在等待該信號量）等。

三、基于信號量的生產(chǎn)者消費者模型

3、1 信號量控制環(huán)形隊列

? 我們之前學(xué)習(xí)了? 生產(chǎn)者消費者問題（條件變量 & 互斥鎖）。之前學(xué)習(xí)的時由阻塞隊列來實現(xiàn)的。通過互斥鎖與條件變量很好的維護了生產(chǎn)者與消費者之前的同步與互斥關(guān)系。那么我么那接下來看看用信號量來維護生產(chǎn)者和消費者之間的同步與互斥關(guān)系的環(huán)形隊列。

3、1、1 空間資源和數(shù)據(jù)資源

? 我們先看下圖：

? 生產(chǎn)者就是要生產(chǎn)數(shù)據(jù)放進(jìn)環(huán)形隊列中去。那么生產(chǎn)者所需要的就是申請環(huán)形隊列空間資源。這個空間的大小我們可以自定義，比如環(huán)形隊列由10個空間資源。

? 消費者就是去環(huán)形隊列拿數(shù)據(jù)。消費者所需要的就是申請數(shù)據(jù)資源。也就是看環(huán)形隊列中是否還有數(shù)據(jù)資源。?

3、1、2 保護原理（二元信號量）

? 我們發(fā)現(xiàn)：生產(chǎn)和消費在隊列為空的時候或者滿的時，可能訪問同一個位置。那我們必須保證生產(chǎn)者生產(chǎn)的數(shù)據(jù)個數(shù)最多不能超過環(huán)形隊列的容量，其次消費者在空的時候不能再拿數(shù)據(jù)。

? 那我們就可以用兩個信號量來很好的維護這兩個角色的需求。生產(chǎn)者對應(yīng)空間資源信號量，消費者對應(yīng)數(shù)據(jù)資源信號量。當(dāng)申請對應(yīng)的信號量資源失敗時，也會進(jìn)入阻塞式等待。直到有信號量資源才會繼續(xù)執(zhí)行。

? 當(dāng)只有單生產(chǎn)單消費時，我們只需要維護好生產(chǎn)與消費的同步與互斥關(guān)系。多生產(chǎn)與多消費時，還需維護生產(chǎn)與生產(chǎn)、消費與消費的互斥關(guān)系。下面我們直接看代碼。

3、2 demo代碼

3、2、1 單生產(chǎn)與單消費

ringQueue.hpp

#include<iostream>
#include<semaphore.h>
#include<unistd.h>
#include<vector>
#include<assert.h>
#include<pthread.h>
#include<time.h>
#include<string.h>


using namespace std;

#include"Task.hpp"
#include"LogTest.hpp"

static const int g_cap=5;

template<class T>
class RingQueue
{
private:
    void P(sem_t& sem)
    {
        int n = sem_wait(&sem);
        assert(n == 0);
        (void)n;
    }

    void V(sem_t& sem)
    {
        int n = sem_post(&sem);
        assert(n == 0);
        (void)n;
    }

public:
    RingQueue(int cap = g_cap)
        :_cap(cap)
    {
        int n = sem_init(&_spaceSem, 0, _cap);
        assert(n == 0);
        n = sem_init(&_dataSem, 0, 0);
        assert(n == 0);

        _queue.resize(_cap);
        _productorStep = _consumerStep = 0;
    }

    // 生產(chǎn)者
    void push(const T& in)
    {
        P(_spaceSem);
        _queue[_productorStep++] = in;
        _productorStep %= _cap;
        V(_dataSem);
    }

    // 消費者
    void pop(T* out)
    {
        P(_dataSem);
        *out = _queue[_consumerStep++];
        _consumerStep %= _cap;
        V(_spaceSem);
    }

    ~RingQueue()
    {
        sem_destroy(&_spaceSem);
        sem_destroy(&_dataSem);
    }
private:
    sem_t _spaceSem; // 生產(chǎn)者——空間資源
    sem_t _dataSem;  // 消費者——數(shù)據(jù)資源
    vector<T> _queue;
    int _cap;
    int _productorStep;
    int _consumerStep;
};

testMain.cpp

#include "ringQueue.hpp"

int myAdd(int x, int y)
{
    return x + y;
}

void* ProductorRoutine(void *arg)
{
    RingQueue<Task> *rq = (RingQueue<Task>*) arg;
    while(true)
    {
        int x = rand() % 10 +1;
        int y = rand() % 100 + 1;
        Task t(x, y, myAdd);
        rq->push(t);
        LogMessage(1,"%s：%d + %d = ?","生產(chǎn)者申請了一個空間", x, y);
        //sleep(1);
    }
}

void* ConsumerRoutine(void* arg)
{
    RingQueue<Task> *rq = (RingQueue<Task>*)arg;
    while(true)
    {
        Task t;
        rq->pop(&t);
        LogMessage(1,"%s：%d + %d = %d","消費者消費了一個數(shù)據(jù)", t.x_, t.y_, t());
        sleep(1);
    }
}
int main()
{
    srand((unsigned int)time(nullptr) ^ 0x666888);
    RingQueue<Task> *rq = new RingQueue<Task>();

    pthread_t c, p;
    pthread_create(&p, nullptr, ProductorRoutine, rq);
    pthread_create(&c, nullptr, ConsumerRoutine, rq);

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

? 通過上述代碼我們發(fā)現(xiàn)：信號量有點類似于互斥鎖+條件變量的結(jié)合，不還是在競爭資源串行訪問嗎？實際上剛開始我們并不知道是先生產(chǎn)，還是先消費。如果先消費，則需等待。如果滿的情況下，生產(chǎn)也需要等待。其他情況下大部分時間都是在并發(fā)執(zhí)行的。生產(chǎn)和消費可同時進(jìn)行。

?3、2、2 多生產(chǎn)多消費

? 當(dāng)多生產(chǎn)和多消費時，我們還需維護生產(chǎn)與生產(chǎn)、消費與消費的互斥關(guān)系。這時候就需要互斥鎖來維護了。代碼如下：

ringQueue.hpp

#pragma once

#include<iostream>
#include<semaphore.h>
#include<unistd.h>
#include<vector>
#include<assert.h>
#include<pthread.h>
#include<time.h>
#include<string.h>


using namespace std;

#include"Task.hpp"
#include"LogTest.hpp"

static const int g_cap=5;

template<class T>
class RingQueue
{
private:
    void P(sem_t& sem)
    {
        int n = sem_wait(&sem);
        assert(n == 0);
        (void)n;
    }

    void V(sem_t& sem)
    {
        int n = sem_post(&sem);
        assert(n == 0);
        (void)n;
    }

public:
    RingQueue(int cap = g_cap)
        :_cap(cap)
    {
        int n = sem_init(&_spaceSem, 0, _cap);
        assert(n == 0);
        n = sem_init(&_dataSem, 0, 0);
        assert(n == 0);

        _queue.resize(_cap);
        _productorStep = _consumerStep = 0;
    }

    // 生產(chǎn)者
    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);
    }
private:
    sem_t _spaceSem; // 生產(chǎn)者——空間資源
    sem_t _dataSem;  // 消費者——數(shù)據(jù)資源
    vector<T> _queue;
    int _cap;
    int _productorStep;
    int _consumerStep;

    pthread_mutex_t _pmutex;
    pthread_mutex_t _cmutex;
};

testMain.cpp

#include "ringQueue.hpp"

int myAdd(int x, int y)
{
    return x + y;
}

void* ProductorRoutine(void *arg)
{
    RingQueue<Task> *rq = (RingQueue<Task>*) arg;
    while(true)
    {
        int x = rand() % 10 +1;
        int y = rand() % 100 + 1;
        Task t(x, y, myAdd);
        rq->push(t);
        LogMessage(1,"%s：%d + %d = ?","生產(chǎn)者申請了一個空間", x, y);
        //sleep(1);
    }
}

void* ConsumerRoutine(void* arg)
{
    RingQueue<Task> *rq = (RingQueue<Task>*)arg;
    while(true)
    {
        Task t;
        rq->pop(&t);
        LogMessage(1,"%s：%d + %d = %d","消費者消費了一個數(shù)據(jù)", t.x_, t.y_, t());
        sleep(1);
    }
}
int main()
{
    srand((unsigned int)time(nullptr) ^ 0x666888);
    RingQueue<Task> *rq = new RingQueue<Task>();

    pthread_t p[4], c[8];
    for(int i = 0; i < 4; i++) pthread_create(p+i, nullptr, ProductorRoutine, rq);
    for(int i = 0; i < 8; i++) pthread_create(c+i, nullptr, ConsumerRoutine, rq);

    for(int i = 0; i < 4; i++) pthread_join(p[i], nullptr);
    for(int i = 0; i < 8; i++) pthread_join(c[i], nullptr);
    delete rq;
    return 0;
}

? 這里又有一個小細(xì)節(jié)：我們在插入或者刪除時，是先加鎖再申請信號量呢，還是先申請信號量再加鎖呢？?首先申請信號量的操作時原子的，這個問題不用擔(dān)心。關(guān)鍵在于插入和刪除的過程。其實先加鎖再申請信號量是肯定可行的，就是先申請信號量再加鎖可以嗎？答案是可以的！申請信號量的本質(zhì)就是在對資源的預(yù)定。只要你申請信號量成功，就一定有資源可用。當(dāng)我們先把信號量申請完，就是把資源先分配給了不同線程。反而會更好一點。后續(xù)各個線程不用競爭信號資源了，只需競爭鎖資源就可以了。文章來源地址http://www.zghlxwxcb.cn/news/detail-719943.html

到了這里，關(guān)于【Linux從入門到精通】信號量（信號量的原理與相關(guān)操作接口）詳解的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！