【Linux】信号量基于环形队列的生产消费模型

信号量

信号量的本质是一个计数器，可以用来衡量临界资源中资源数量多少

信号量的PV操作

P操作：申请信号量称为P操作，P操作的本质就是让计数器减1。

V操作：释放信号量称为V操作，V操作的本质就是让计数器加1

POSIX信号量相关的接口函数

初始化信号量

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

sem：需要初始化的信号量。

pshared：0表示线程间共享，非0表示进程间共享。

value：信号量的初始值（特定资源的初始数量）。

等待信号量（P操作)

int sem_wait(sem_t *sem);

发布信号量（V操作)

int sem_post(sem_t *sem);

销毁信号量

int sem_destroy(sem_t *sem);

基于环形队列的生产消费模型代码

RingQueue.hpp

#pragma once

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

static const int gcap = 5;

template<class T>
class 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<T> _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 <iostream>
#include <string>
#include <cstdio>
#include <functional>

class Task
{
    using func_t = std::function<int(int,int,char)>;
    // typedef std::function<int(int,int)> 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 << "div zero error!" << std::endl;
            result = -1;
        }
        else
            result = x / y;
    }
        break;
    case '%':
    {
        if (y == 0)
        {
            std::cerr << "mod zero error!" << std::endl;
            result = -1;
        }
        else
            result = x % y;
    }
        break;
    default:
        // do nothing
        break;
    }

    return result;
}

main.cc

#include "RingQueue.hpp"
#include "Task.hpp"
#include <pthread.h>
#include <ctime>
#include <cstdlib>
#include <sys/types.h>
#include <unistd.h>

std::string SelfName()
{
    char name[128];
    snprintf(name, sizeof(name), "thread[0x%x]", pthread_self());
    return name;
}

void *ProductorRoutine(void *rq)
{
    RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
    while(true)
    {
        int x = rand() % 10;
        int y = rand() % 5;
        char op = oper[rand()%oper.size()];
        Task t(x, y, op, mymath);
        // 生产任务
        ringqueue->Push(t);
        // 输出提示
        std::cout <<  SelfName() << ", 生产者派发了一个任务: " << t.toTaskString() << std::endl;
        sleep(1);
    }
}

void *ConsumerRoutine(void *rq)
{
    RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);

    while(true)
    {
        Task t;
        //消费任务
        ringqueue->Pop(&t);
        std::string result = t(); 
        std::cout <<  SelfName() << ", 消费者消费了一个任务: " << result << std::endl;
    }
}

int main()
{
    srand((unsigned int)time(nullptr) ^ getpid() ^ pthread_self() ^ 0x71727374);
    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;
}