信號量

信號量的本質是一個計數器

首先一份公共資源在實際情況中可能存在不同的線程可以并發訪問不同的區域。因此當一個線程想要訪問臨界資源時，可以先申請信號量，只要信號量申請成功就代表著這個線程在未來一定能訪問臨界資源的一部分。而這個申請信號量的本質就是對臨界資源中特定的小塊資源進行預定機制

所有的線程都能看到信號量，也就代表著信號量的本身就是公共資源。如果信號量申請失敗，說明該線程并不滿足條件變量則線程進入阻塞等待。

信號量的操作可以稱為 PV操作，申請信號量成功則信號量的值減1（P），歸還資源后信號量的值加1（V）。在這過程中一定要保證原子性

信號量的接口

信號量的類型為：sem_t

初始化

定義一個信號量并初始化：

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

參數一為信號量的地址

參數二如果設為0則表示線程間共享，非0則表示進程間共享

參數三為信號量的初始值，也就是這個計數器的最大值

初始化成功返回0

銷毀

int sem_destroy(sem_t *sem);

參數為信號量的地址

等待信號量

也就是申請信號量，等待成功則信號量的值減1

int sem_wait(sem_t *sem);

參數為信號量的地址

發布信號量

也就是歸還信號量，發布成功則信號量的值加1

int sem_post(sem_t *sem);

參數為信號量的地址

環形隊列

將生產者消費者模型放到環形隊列里該如何理解呢？

首先環形隊列可以理解為就是指一個循環的數組。而對于生產者和消費者而言，生產者負責往這個數組中放入數據，消費者負責從這個數組中拿取數據。

對于生產者而言：只有環形隊列中有空的空間時才可以放數據

對于消費者而言：環形隊列中必須要有不為空的空間時才能拿數據

那么肯定是生產者放入數據后消費者才能拿數據，因此在喚醒隊列中消費者永遠都不能超過生產者

假如生產者和消費者都指向了同一塊空間時，要么隊列滿了，要么隊列為空

如果隊列滿了，那么消費者就必須要先運行拿走數據后生產者才可以運行

如果隊列為空，那么生產者必須放入數據后消費者才能運行拿數據

如果為其他情況，說明生產者和消費者所指向的空間是不一樣的

結合信號量設計模型

那么根據上述的消費者和生產者不同的特性，結合信號量為兩者設計條件

對于生產者而言看中的是隊列中剩余的空間，那么可以給生產者設置信號量為隊列的總容量

對于消費者而言看中的是隊列中不為空的空間，則可以給消費者初始的信號量值設為0

那么對于生產過程而言：

1. 首先生產者去申請信號量也就是 P 操作
2. 申請成功則往下繼續生產操作的執行，申請失敗則阻塞等待
3. 執行完生產操作后，V操作，注意此時的V操作并不是對生產者的信號量操作，而是對消費者的信號量操作。因為此時隊列中已經有了不為空的空間，所以消費者的信號量就可以進行加1操作，這樣消費者才能申請成功信號量

對于消費過程而言:

1. 消費者申請信號量
2. 申請成功則往下繼續消費操作的執行，失敗則阻塞等待
3. 執行完消費操作后，V操作，注意此時V操作是對生產者的信號量操作，因為消費完成后，隊列里就多了一個空的空間，生產者的信號量就可以進行加1操作

那么對于生產者和消費者的位置其實就是隊列中的下標，并且一定是有兩個下標，如果隊列為空為滿，那么兩者的位置相同

實現基于環形隊列的生產者消費者模型

Task.hpp

#include <iostream>
#include <string>
#include <functional>
#include <cstdio>

// 負責計算的任務類
class CPTask
{
    // 調用的計算方法，根據傳入的字符參數決定
    typedef std::function<int(int, int, char)> func_t;

public:
    CPTask()
    {
    }

    CPTask(int x, int y, char op, func_t func)
        : _x(x), _y(y), _op(op), _func(func)
    {
    }

    // 實現傳入的函數調用
    std::string operator()()
    {
        int count = _func(_x, _y, _op);

        // 將結果以自定義的字符串形式返回
        char res[2048];
        snprintf(res, sizeof res, "%d %c %d = %d", _x, _op, _y, count);
        return res;
    }

    // 顯示出當前傳入的參數
    std::string tostring()
    {
        char res[1024];
        snprintf(res, sizeof res, "%d %c %d = ", _x, _op, _y);
        return res;
    }

private:
    int _x;
    int _y;
    char _op;
    func_t _func;
};

// 負責計算的任務函數
int Math(int x, int y, char c)
{
    int count;
    switch (c)
    {
    case '+':
        count = x + y;
        break;
    case '-':
        count = x - y;
        break;
    case '*':
        count = x * y;
        break;
    case '/':
    {
        if (y == 0)
        {
            std::cout << "div zero" << std::endl;
            count = -1;
        }
        else
            count = x / y;
        break;
    }
    default:
        break;
    }

    return count;
}

class SCTask
{
    // 獲取保存數據的方法
    typedef std::function<void(std::string)> func_t;

public:
    SCTask()
    {
    }

    SCTask(const std::string &str, func_t func)
        : _str(str), _func(func)
    {
    }

    void operator()()
    {
        _func(_str);
    }

private:
    std::string _str;
    func_t _func;
};

RingQueue.hpp

#pragma once

#include <iostream>
#include <pthread.h>
#include <vector>
#include <semaphore.h>
#include <string>
#include <ctime>
#include <unistd.h>
#include <cassert>

using namespace std;

template <class T>
class RingQueue
{
public:
    RingQueue(const int size = 10)
        : _size(size)
    {
        // 初始化信號量以及數組大小
        // 生產者的信號量初始為數組大小
        // 消費者的信號量初始為0
        _Rqueue.resize(_size);
        assert(sem_init(&_ps, 0, _size) == 0);
        assert(sem_init(&_cs, 0, 0) == 0);
    }

    void push(const T &in)
    {
        // 生產者申請信號量
        assert(sem_wait(&_ps) == 0);

        // 為了模擬環形，下標需要模等于數組的大小
        // 保證下標不越界
        _Rqueue[_Pindex++] = in;
        _Pindex %= _size;

        // 消費者發布信號量
        assert(sem_post(&_cs) == 0);
    }

    void pop(T *out)
    {
        // 消費者申請信號量
        assert(sem_wait(&_cs) == 0);

        // 為了模擬環形，下標需要模等于數組的大小
        // 保證下標不越界
        *out = _Rqueue[_Cindex++];
        _Cindex %= _size;

        // 生產者發布信號量
        assert(sem_post(&_ps) == 0);
    }

    ~RingQueue()
    {
        sem_destroy(&_ps);
        sem_destroy(&_cs);
    }

private:
    vector<T> _Rqueue;
    int _size;       // 數組的容量大小 
    sem_t _ps;       // 生產者的信號量
    sem_t _cs;       // 消費者的信號量
    int _Pindex = 0; // 生產者的下標
    int _Cindex = 0; // 消費者的下標
};

main.cc

#include "RingQueue.hpp"
#include "Task.hpp"

void *Pplay(void *rp)
{
    RingQueue<CPTask> *rq = (RingQueue<CPTask> *)rp;

    while (1)
    {
        string s("+-*/");
        // 隨機產生數據插入
        int x = rand() % 100 + 1;
        int y = rand() % 100 + 1;
        // 隨機提取+-*/
        int i = rand() % s.size();
        // 定義好實現類的對象
        CPTask c(x, y, s[i], Math);

        // 將整個對象插入到計算隊列中
        rq->push(c);
        cout << "生產數據完成: " << c.tostring() << endl;
        sleep(1);
    }
}

void *Cplay(void *cp)
{
    RingQueue<CPTask> *rq = (RingQueue<CPTask> *)cp;

    while (1)
    {
        // 隊列拿出數據
        CPTask c;
        rq->pop(&c);
        // 調用拿出的對象獲取最終的結果
        string s = c();

        cout << "消費完成，取出的數據為: " << s << endl;
        sleep(2);
    }
}

int main()
{
    srand(time(nullptr));

    RingQueue<CPTask> *rq = new RingQueue<CPTask>();

    pthread_t p, c;
    pthread_create(&p, nullptr, Pplay, (void *)rq);
    pthread_create(&c, nullptr, Cplay, (void *)rq);

    pthread_join(p, nullptr);
    pthread_join(c, nullptr);

    delete rq;

    return 0;
}

效果

對于多生產多消費的情況

如果現在有多個生產和多個消費，那么就要保證臨界資源的安全，這時候就得加鎖。

加鎖可以實現在申請信號量之后，因為申請信號量實際上就是一個原子性的操作，并不會因為多線程而導致沖突。當線程申請好了各自的信號量之后再往下運行加鎖，就不用讓所有線程都等待在加鎖前。而解鎖同樣可以在發布信號量之后。