【linux线程(三)】生产者消费者模型详解(多版本)

#pragma once
#include<iostream>
#include<pthread.h>
using namespace std;
class Mutex //封装pthread库的锁
{
public:
    Mutex(pthread_mutex_t* pmtx)
        :_pmtx(pmtx)
    {}
    void lock()
    {
        pthread_mutex_lock(_pmtx);
    }
    void unlock()
    {
        pthread_mutex_unlock(_pmtx);
    }
    ~Mutex()
    {
        delete _pmtx;
        _pmtx = nullptr;
    }   
private:
    pthread_mutex_t* _pmtx;
};
//利用RAII思想,用对象的生命周期控制资源
class LockGuard//封装了锁后,出了作用域会自动调用析构函数,解锁!
{
public:
    LockGuard(pthread_mutex_t* mtx)
        :_mtx(mtx)
    {
        _mtx.lock();
    }
    ~LockGuard()
    {
        _mtx.unlock();
    }
private:
    Mutex _mtx;
};

#pragma once
#include<iostream>
#include<queue>
#include<pthread.h>
#include"LockGuard.cpp"
using namespace std;
#define INIT_MTX(mtx) pthread_mutex_init(&mtx,nullptr)
#define INIT_COND(cond) pthread_cond_init(&cond,nullptr)
#define DESTORY_MTX(mtx) pthread_mutex_destroy(&mtx)
#define DESTORY_COND(cond) pthread_cond_destroy(&cond)
template<class T>
class BlockQueue
{
public:
    BlockQueue(size_t capacity = 5)
        :_capacity(capacity)
    {
        INIT_MTX(_mtx);
        INIT_COND(_condisempty);
        INIT_COND(_condisfull);
    }
    ~BlockQueue()
    {
        DESTORY_MTX(_mtx);
        DESTORY_COND(_condisempty);
        DESTORY_COND(_condisfull);
    }
    void push(const T& in)
    {
        //1. 加锁
        pthread_mutex_lock(&_mtx);
        //2. 检测临界资源是否满足访问条件,不满足就利用条件变量让它等待
        while(isfull())
        {
            cout<<"队列已满,push等待中..."<<endl;
            //我是抱着锁去等待的,所以即使消费者去pop了,没有锁也只能阻塞等待,就造成死循环了!
            //pthread_cond_wait的第二个参数是一把锁,当成功调用wait后,传入的锁会自动释放!
            //被唤醒时,还是在加锁和解锁之间,wait函数会自动帮线程获取锁
            // 1. wait函数可能会调用失败
            // 2. wait函数可能存在伪唤醒的情况
            // 所以不能使用if语句来判断,而是用while,醒来后再判断一下是不是真的唤醒
            pthread_cond_wait(&_condisfull,&_mtx);//若队列满了,就去isfull变量中等待
            cout<<"等待成功,继续执行..."<<endl;
        }
        //3. 访问临界资源
        _bq.push(in);
        //5. 生产完后唤醒消费者(满足一定条件再唤醒)
        if(_bq.size() >= _capacity/2)
            pthread_cond_signal(&_condisempty);
        //4. 解锁
        pthread_mutex_unlock(&_mtx);
    }
    void pop(T* out)
    {
        //1. 加锁
        pthread_mutex_lock(&_mtx);
        //2. 检测临界资源是否满足访问条件,不满足就利用条件变量让它等待
        while(isempty())
        {
            cout<<"队列为空,pop等待中..."<<endl;
            pthread_cond_wait(&_condisempty,&_mtx);
            cout<<"等待成功,继续执行..."<<endl;
        }
        //3. 访问临界资源
        *out = _bq.front();
        _bq.pop();
        //4. 解锁
        pthread_mutex_unlock(&_mtx);
        //5. 消费完后唤醒生产者
        pthread_cond_signal(&_condisfull);
    }
    bool isempty()
    {
        return _bq.size() == 0;
    }
    bool isfull()
    {
        return _bq.size() ==_capacity;
    }
private:
    queue<T> _bq;
    size_t _capacity;  //阻塞队列容量上限
    pthread_mutex_t _mtx;  //通过互斥锁保证队列push/pop时,不会被pop/push(正在生产时就来消费)
    //push或pop时,怎么知道队列是不是满或空呢?通过条件变量来同步信息
    pthread_cond_t _condisempty; //用它来表示阻塞队列是否为空的条件
    pthread_cond_t _condisfull; //用它来表示阻塞队列是否为满的条件
};

#pragma once
#include<semaphore.h>
using namespace std;
class Sem
{
public:
    Sem(int value)
    {
        sem_init(&_sem,0,value);
    }
    ~Sem()
    {
        sem_destroy(&_sem);
    }
    void p() //PV操作分别代表信号量减一和信号量加一
    {
        sem_wait(&_sem);
    }
    void v()
    {
        sem_post(&_sem);
    }
private:
    sem_t _sem;
};

#pragma once
#include<iostream>
#include<pthread.h>
#include<vector>
#include<unistd.h>
#include<semaphore.h>
using namespace std;
#include"Sem.hpp"
const int g_num = 5;
template<class T>
class RingQueue
{
public:
    RingQueue(int num = g_num)
        :_num(num)
        ,_rq(num)
        ,_pushindex(0)
        ,_popindex(0)
        ,_space_sem(num)//最开始空间资源是满的
        ,_data_sem(0)//最开始数据资源是空的
    {
        pthread_mutex_init(&pushlock,nullptr);
        pthread_mutex_init(&poplock,nullptr);
    }
    ~RingQueue()
    {
        pthread_mutex_destroy(&pushlock);
        pthread_mutex_destroy(&poplock);
    }
    void push(const T& in)//生产者进行,关心空间资源,生产者们的临界资源是pushindex下标
    {
        _space_sem.p();
        pthread_mutex_lock(&pushlock);//对生产者们进行加锁
        _rq[_pushindex++] = in;
        _pushindex %= _num;
        pthread_mutex_unlock(&pushlock);
        _data_sem.v();
    }
    void pop(T* out)//消费者进行,关心数据资源，消费者们的临界资源也是popindex下标
    {
        //一般先申请信号量，再进行加锁
        _data_sem.p();
        pthread_mutex_lock(&poplock);//对消费者们进行加锁
        //竞争成功的生产者线程只有一个
        *out = _rq[_popindex++];
        _popindex %= _num;
        pthread_mutex_unlock(&poplock);
        _space_sem.v();
    }
private:
    vector<T> _rq;//环形队列可用数组表示
    size_t _num;//总共的元素个数
    size_t _pushindex = 0;//push的下标
    size_t _popindex = 0;//pop的下标
    Sem _space_sem;
    Sem _data_sem;
    pthread_mutex_t pushlock;//生产者的锁
    pthread_mutex_t poplock;//消费者的锁
};