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

这篇博客的重点在于代码实现，理论部分请看CSDN​​​​​​​

一、单生产单消费

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

二、多生产多消费

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;       // 消费者的位置(数组下标)
    pthread_mutex_t _pmutex; // 生产者
    pthread_mutex_t _cmutex; // 消费者
};

(1) 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;

        pthread_mutex_init(&_pmutex, nullptr);
        pthread_mutex_init(&_cmutex, nullptr);
    }

(2) ~RingQueue()

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

        pthread_mutex_destroy(&_pmutex);
        pthread_mutex_destroy(&_cmutex);
    }

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

    // 生产者
    void Push(const T &in)
    {
        //? 代码有没有优化的可能
        // 你认为：现加锁，后申请信号量;还是现申请信号量，在加锁？
        // pthread_mutex_lock(&_pmutex);
        // 申请到了信号量，意味着我一定能进行正常的生产
        P(_spaceSem);
        pthread_mutex_lock(&_pmutex);
        _queue[_productorStep++] = in;
        _productorStep %= _cap;
        pthread_mutex_unlock(&_pmutex);
        V(_dataSem);
        // pthread_mutex_unlock(&_pmutex);
    }

(4) void Pop(T *out);

    // 消费者
    void Pop(T *out)
    {
        // 你认为：现加锁，后申请信号量;还是现申请信号量，在加锁？
        // pthread_mutex_lock(&_cmutex);
        P(_dataSem);
        pthread_mutex_lock(&_cmutex);
        *out = _queue[_consumerStep++];
        _consumerStep %= _cap;
        pthread_mutex_unlock(&_cmutex);
        V(_spaceSem);
        // pthread_mutex_unlock(&_cmutex);
    }

2.上层调用逻辑

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

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

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 << SelName() << ",生产者派发了一个任务：" << 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[4], c[8];
    for (int i = 0; i < 4; i++)
        pthread_create(p + i, nullptr, ProductorRoutine, rq);
    for (int i = 0; i < 4; 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 < 4; i++)
        pthread_join(c[i], nullptr);
    delete rq;
    return 0;
}