Linux操作系统7- 线程同步与互斥7（RingQueue环形队列生产者消费者模型改进）

上篇文章：Linux操作系统7- 线程同步与互斥6（POSIX信号量与环形队列生产者消费者模型）-CSDN博客

本篇代码仓库：myLerningCode/l36 · 橘子真甜/Linux操作系统与网络编程学习 - 码云 - 开源中国 (gitee.com)

目录

一. 单生产单消费单保存模型

1.1 RingQueue.hpp

1.2 Task.hpp

1.3 MainPC.cpp

1.4 测试

二. 多生产多消费模型        

2.1 分析与代码 

2.2 多生产多消费的意义

一. 单生产单消费单保存模型

        通过RingQueue可以实现生产者消费者之间的协同工作，如果现在想要将消费者的输出结果保存在文件中应该怎么办？

        可以定义两个环形队列，三个线程。让消费者充当第二个队列的生产者。

代码如下：

1.1 RingQueue.hpp

        直接使用上篇文件的代码即可。然后我们需要新增一个类，这个类中包含两个环形队列用于消费者同时访问两个队列

#pragma once
#include <iostream>
#include <vector>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
const int gnum = 10; // 阻塞队列的最大容量

template <class T>
class RingQueue
{
private:
    void P(sem_t &sem)
    {
        // P操作，申请信号量sem--
        int n = sem_wait(&sem);
        if (n < 0)
            std::cerr << "P操作失败" << std::endl;
    }

    void V(sem_t &sem)
    {
        // V操作，释放信号量,sem++
        int n = sem_post(&sem);
        if (n < 0)
            std::cerr << "V操作失败" << std::endl;
    }

public:
    RingQueue(const int &maxnum = gnum) : _maxnum(maxnum), _ringQueue(gnum)
    {
        // 在构造函数中完成信号量的初始化
        int n = sem_init(&_spaceSem, 0, maxnum);
        if (n < 0)
            std::cerr << "spaceSem信号量初始化失败" << std::endl;
        n = sem_init(&_dataSem, 0, 0);
        if (n < 0)
            std::cerr << "dataSem信号量初始化失败" << std::endl;

        _producerStep = _consumerStep = 0;
    }

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

    // 生产者插入数据
    void push(const T &in)
    {
        // 申请空间资源
        P(_spaceSem);

        // 插入数据
        _ringQueue[_producerStep++] = in;
        _producerStep %= _maxnum;

        // 释放一个数据资源
        V(_dataSem);
    }

    // 消费者获取数据
    void pop(T *out)
    {
        // 申请数据资源
        P(_dataSem);

        // 插入数据
        *out = _ringQueue[_consumerStep++];
        _consumerStep %= _maxnum;

        // 释放一个空间资源
        V(_spaceSem);
    }

private:
    std::vector<T> _ringQueue; // 使用数组来实现环形队列
    size_t _maxnum;

    sem_t _spaceSem; // 生产者空间资源信号量
    sem_t _dataSem;  // 消费者数据资源信号量

    int _producerStep; // 生产者下标
    int _consumerStep; // 消费者下标
};

1.2 Task.hpp

        需要新增一个保存者的任务

#pragma once
#include <iostream>
#include <cstdio>
#include <functional>

class CalTask
{
    using func_t = std::function<int(int, int, char)>; // func是一个函数
    // typedef std::function<int(int,int)> func;
public:
    CalTask() {}

    CalTask(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) - 1, "%d %c %d = %d", _x, _op, _y, result);
        return buffer;
    }

    // 返回任务操作的结果
    std::string toString()
    {
        char buffer[1024];
        snprintf(buffer, sizeof(buffer) - 1, "%d %c %d = ?", _x, _op, _y);
        return buffer;
    }

private:
    int _x;
    int _y;
    char _op;         // 操作的任务的id
    func_t _callback; // 调用的函数
};

class SaveTask
{
    typedef std::function<void(const std::string &)> func_t;

public:
    SaveTask() {}
    SaveTask(const std::string &message, func_t func)
        : _message(message), _callback(func) {}
    void operator()()
    {
        _callback(_message);
    }

private:
    std::string _message; // 保存的信息
    func_t _callback;     // 将信息写入文件中
};

const std::string oper = "+-*/%";
int my_math(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" << std::endl;
            return -1;
        }
        else
        {
            result = x / y;
        }
        break;
    }
    case '%':
        if (y == 0)
        {
            std::cerr << "moved zero" << std::endl;
            return -1;
        }
        else
        {
            result = x % y;
        }
        break;
    default:
        break;
    }
    return result;
}

void Save(const std::string &message)
{
    const std::string task_pwd = "./log.txt";
    FILE *fp = fopen(task_pwd.c_str(), "a+");
    if (nullptr == fp)
    {
        std::cerr << "saver open error" << std::endl;
        return;
    }
    fputs(message.c_str(), fp);
    fputc('\n', fp);

    fclose(fp);
}

1.3 MainPC.cpp

#include "RingQueue.hpp"
#include "Task.hpp"
#include <iostream>

// 生产者
void *ProductorRoutine(void *args)
{
    RingQueue<CalTask> *cal_rq = (static_cast<RingQueues<CalTask, SaveTask> *>(args))->_cal_rq;
    while (true)
    {
        int x = rand() % 20;
        int y = rand() % 50;
        const char op = oper[rand() % oper.size()];
        CalTask ct(x, y, op, my_math);
        cal_rq->push(ct);
        std::cout << "生产者生产任务:" << ct.toString() << " 并传递给消费者完成" << std::endl;
    }
}

// 消费者
void *ConsumerRoutine(void *args)
{
    RingQueue<CalTask> *cal_rq = (static_cast<RingQueues<CalTask, SaveTask> *>(args))->_cal_rq;
    RingQueue<SaveTask> *save_rq = (static_cast<RingQueues<CalTask, SaveTask> *>(args))->_save_rq;

    while (true)
    {
        CalTask ct;
        cal_rq->pop(&ct);
        std::string result = ct();
        std::cout << "消费者实现任务:" << result << " 实现完成！" << std::endl;
        SaveTask st(result, Save);
        save_rq->push(st);
        std::cout << "消费者传递任务:" << result << " 给保存者完成！" << std::endl;
    }
}

// 保存者
void *SaverRoutine(void *args)
{
    RingQueue<SaveTask> *save_rq = (static_cast<RingQueues<CalTask, SaveTask> *>(args))->_save_rq;
    while (true)
    {
        SaveTask st;
        save_rq->pop(&st);
        st();
        std::cout << "保存者保存任务完成！" << std::endl;
    }
}

void test1()
{
    RingQueues<CalTask, SaveTask> *rqs = new RingQueues<CalTask, SaveTask>;
    rqs->_cal_rq = new RingQueue<CalTask>();
    rqs->_save_rq = new RingQueue<SaveTask>();

    pthread_t p, c, s;
    pthread_create(&p, nullptr, ProductorRoutine, rqs);
    pthread_create(&c, nullptr, ConsumerRoutine, rqs);
    pthread_create(&s, nullptr, SaverRoutine, rqs);

    pthread_join(p, nullptr);
    pthread_join(c, nullptr);
    pthread_join(s, nullptr);
    delete rqs;
}

int main()
{
    srand((unsigned int)time(0) ^ getpid() ^ pthread_self());
    test1();
    return 0;
}

1.4 测试

        运行结果如下：

二. 多生产多消费模型        

2.1 分析与代码 

        RingQueue环形队列可以保证单个生产者和单个消费者之间的同步与互斥，如果现在有多个生产者和多个消费者的话。如何保证生产者之间的互斥？消费者者之间的互斥？

        阻塞队列中，我们通过加锁的方式让同一时刻只能有一个生产者线程进入临界区或者一个消费者进入临界区。

        而环形队列中， 通过信号量保证了生产者消费者之间的同步与互斥。如果想要保证消费者与消费者之间的互斥，生产者与生产者之间的互斥，也需要加锁保护

        在RingQueue中添加两个成员变量，一个生产者互斥锁，一个消费者互斥锁。同时需要在构造函数中完成锁的初始化，析构函数中完成锁的销毁。

        并且在push函数中加生产者锁，在pop函数中加消费者锁。以实现生产者与生产者之间的互斥和消费者与消费者之间的互斥。（本质是防止多个线程同时访问导致生产者下标或者消费者下标出现数据错误） 

代码如下：

#pragma once
#include <iostream>
#include <vector>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
const int gnum = 10; // 阻塞队列的最大容量

template <class T>
class RingQueue
{
private:
    void P(sem_t &sem)
    {
        // P操作，申请信号量sem--
        int n = sem_wait(&sem);
        if (n < 0)
            std::cerr << "P操作失败" << std::endl;
    }

    void V(sem_t &sem)
    {
        // V操作，释放信号量,sem++
        int n = sem_post(&sem);
        if (n < 0)
            std::cerr << "V操作失败" << std::endl;
    }

public:
    RingQueue(const int &maxnum = gnum) : _maxnum(maxnum), _ringQueue(gnum)
    {
        // 在构造函数中完成信号量的初始化
        int n = sem_init(&_spaceSem, 0, maxnum);
        if (n < 0)
            std::cerr << "spaceSem信号量初始化失败" << std::endl;
        n = sem_init(&_dataSem, 0, 0);
        if (n < 0)
            std::cerr << "dataSem信号量初始化失败" << std::endl;

        _producerStep = _consumerStep = 0;

        // 初始化锁
        pthread_mutex_init(&_pmtx, nullptr);
        pthread_mutex_init(&_cmtx, nullptr);
    }

    ~RingQueue()
    {
        // 销毁信号量与互斥锁
        sem_destroy(&_spaceSem);
        sem_destroy(&_dataSem);

        //
        pthread_mutex_destroy(&_cmtx);
        pthread_mutex_destroy(&_pmtx);
    }

    // 生产者插入数据
    void push(const T &in)
    {
        // 生产者加锁，保证生产者与生产者之间的互斥
        pthread_mutex_lock(&_pmtx);
        // 申请空间资源
        P(_spaceSem);

        // 插入数据
        _ringQueue[_producerStep++] = in;
        _producerStep %= _maxnum;

        // 释放一个数据资源
        V(_dataSem);
        // 解锁
        pthread_mutex_unlock(&_pmtx);
    }

    // 消费者获取数据
    void pop(T *out)
    {
        // 消费者加锁
        pthread_mutex_lock(&_cmtx);
        // 申请数据资源
        P(_dataSem);

        // 插入数据
        *out = _ringQueue[_consumerStep++];
        _consumerStep %= _maxnum;

        // 释放一个空间资源
        V(_spaceSem);

        // 解锁
        pthread_mutex_unlock(&_cmtx);
    }

private:
    std::vector<T> _ringQueue; // 使用数组来实现环形队列
    size_t _maxnum;

    sem_t _spaceSem; // 生产者空间资源信号量
    sem_t _dataSem;  // 消费者数据资源信号量

    int _producerStep; // 生产者下标
    int _consumerStep; // 消费者下标

    pthread_mutex_t _pmtx;
    pthread_mutex_t _cmtx;
};

MainPC.cpp

#include <iostream>
#include <memory>
#include <string>

#include <unistd.h>
#include <pthread.h>
#include "RingQueue.hpp"
#include "Task.hpp"

const std::string OP = "+-*/%";
void *producer(void *args)
{
    // 获取交易场所 - 阻塞队列
    RingQueue<CalTask> *rq = static_cast<RingQueue<CalTask> *>(args);
    while (true)
    {
        int x = rand() % 100;
        int y = rand() % 100;
        char op = OP[rand() % OP.size()];
        // 打印日志
        printf("生产者生产的数据:%d %c %d 并交给消费者计算\n", x, op, y);
        CalTask ct(x, y, op, my_math);
        rq->push(ct);
    }

    return nullptr;
}

void *consumer(void *args)
{
    // 获取交易场所 - 生产消费阻塞队列，消费保存阻塞队列
    RingQueue<CalTask> *rq = static_cast<RingQueue<CalTask> *>(args);

    while (true)
    {
        // 获取任务计算
        CalTask ct;
        rq->pop(&ct);
        std::string result = ct();
        std::cout << "消费者获取数据并计算:" << result << std::endl;
    }
    return nullptr;
}

int main()
{
    srand((unsigned int)time(0) ^ getpid());
    // 建立任务队列和保存队列
    RingQueue<CalTask> *rq = new RingQueue<CalTask>;

    pthread_t c[3], p[3];
    pthread_create(&c[0], nullptr, consumer, (void *)rq);
    pthread_create(&c[1], nullptr, consumer, (void *)rq);
    pthread_create(&c[2], nullptr, consumer, (void *)rq);

    pthread_create(&p[0], nullptr, producer, (void *)rq);
    pthread_create(&p[1], nullptr, producer, (void *)rq);
    pthread_create(&p[2], nullptr, producer, (void *)rq);

    pthread_join(c[0], nullptr);
    pthread_join(c[1], nullptr);
    pthread_join(c[2], nullptr);

    pthread_join(p[0], nullptr);
    pthread_join(p[1], nullptr);
    pthread_join(p[2], nullptr);
    delete rq;
    return 0;
}

测试结果如下：

可以看到，长时间没有出错

如果将加锁解锁操作进行注释:

会出现段错误，因为访问了非法内存

2.2 多生产多消费的意义

        与阻塞队列BlockQueue一样，多生产多消费的时候。生产线程生产数据可能需要很多时间，一个生产者生产者访问环形队列的时候不妨碍其他线程生产自己的资源。一个消费者访问环形队列拿数据的时候不妨碍其他消费者拿到数据进行处理