【操作系统】基于环形队列的生产消费模型-一、单生产单消费

1.环形队列的实现

单生产单消费的情况下，我们只需要维护生产者和消费者之间的互斥和同步关系即可

将环形队列封装成一个类：首先给出整体框架，接着会说明每一个类内函数的实现

#pragma once

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

// 环形队列的默认大小
static const int gcap = 5;

// 设置成模版类型，队列中可以放任意类型的数据
template <class T>
class RingQueue
{
private:
    // 封装系统调用sem_wait
    void P(sem_t &sem);
    // 封装系统调用sem_post
    void V(sem_t &sem);

public:
    RingQueue()
    ~RingQueue()

    // 生产者向队列里放数据
    void Push(const T &in);
    // 消费者从队列中取数据
    void Pop(T *out);

private:
    std::vector<T> _queue; // 数组模拟环形队列
    int _cap;              // 环形队列的容量
    sem_t _spaceSem;       // 生产者 -> 空间资源
    sem_t _dataSem;        // 消费者 -> 数据资源
    int _producerStep;     // 生产者的位置(数组下标)
    int _consumerStep;     // 消费者的位置(数组下标)
};

(1) void P(sem_t &sem);

封装系统调用sem_wait，便于在push和pop中使用

void P(sem_t &sem)
{
    int n = sem_wait(&sem);
    assert(n == 0);
    (void)n;
}

(2) void V(sem_t &sem);

封装系统调用sem_post，便于在push和pop中使用

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

(3) RingQueue()

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;
}

(4) ~RingQueue()

~RingQueue()
{
    sem_destroy(&_spaceSem);
    sem_destroy(&_dataSem);
}

(5) void Push(const T &in);

void Push(const T &in)
{
    // 生产者要了一个空间，空间信号量--
    P(_spaceSem);
    // 把数据放进队列
    _queue[_producerStep++] = in;
    // 维护环状结构
    _producerStep %= _cap;
    // 队列新增了数据，数据信号量++
    V(_dataSem);
}

(6) void Pop(T *out);

    // 消费者从队列中取数据

void Pop(T *out)
{
    // 消费者拿了一个数据，数据信号量--
    P(_dataSem);
    // 把数据拿出队列
    *out = _queue[_consumerStep++];
    // 维护环状结构
    _consumerStep %= _cap;
    // 队列多出了空间，空间信号量++
    V(_spaceSem);
}

2.上层调用逻辑

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

void *ProductorRoutine(void *rq)
{
    RingQueue<int> *ringqueue = static_cast<RingQueue<int> *>(rq);
    // RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
    while (true)
    {
        // // version1
        // int data = rand() % 10 + 1;
        // ringqueue->Push(data);
        // std::cout << "生产完成，生产的数据是：" << data << std::endl;
        // sleep(1);

        // version2
        // 构建/获取 任务 -- 花费时间
        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 << "生产者派发了一个任务：" << t.toTaskString() << std::endl;

        sleep(1);
    }
}

void *ConsumerRoutine(void *rq)
{
    // RingQueue<int> *ringqueue = static_cast<RingQueue<int> *>(rq);
    RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq);
    while (true)
    {
        // // version1
        // int data;
        // ringqueue->Pop(&data);
        // std::cout << "消费完成，消费的数据是：" << data << std::endl;

        // version2
        Task t;
        // 消费任务 -- 花费时间
        ringqueue->Pop(&t);
        std::cout << SelName() << ",消费者消费了一个任务：" << t() << std::endl;
    }
}

int main()
{
    srand((unsigned int)time(nullptr) ^ getpid());

    RingQueue<int> *rq = new RingQueue<int>();
    // RingQueue<Task> *rq = new RingQueue<Task>();

    // 单生产，单消费
    pthread_t p, c;
    pthread_create(p + i, nullptr, ProductorRoutine, rq);
    pthread_create(c + i, nullptr, ConsumerRoutine, rq);

    pthread_join(p[i], nullptr);
    pthread_join(c[i], nullptr);
    delete rq;

    return 0;
}