基于BlockQueue的生产消费模型及Linux中的信号量

news2024/11/28 4:27:35

基于BlockQueue的生产消费模型

Task.hpp

#pragma once

#include<cstdio>
#include<iostream>
#include<string>
#include<functional>

using namespace std;
class CalTask
{
    using func_t=function<int(int,int,char)>;
    //typedef function<int(int,int)> func_t;
public:
    CalTask()
    {}
    CalTask(int x,int y,char op,func_t func):_x(x),_y(y),_op(op),_callback(func)
    {}
    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;
    }
    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;
};

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)
            {
                cerr<<"div zero error!!!"<<endl;
                result=-1;
            }
            else
                result=x/y;
        }
        break;
        case '%':
        {
            if(y==0)
            {
                cerr<<"mod zero error!!!"<<endl;
                result=-1;
            }
            else
                result=x%y;
        }          
        break;
        default:
        break;
    }
    return result;
}
class SaveTask
{
    typedef function<void(const string&)> func_t;
public:
    SaveTask()
    {}
    SaveTask(const string &message,func_t func)
    :_message(message),_func(func)
    {}
    void operator()()
    {
        _func(_message);
    }
private:
    string _message;
    func_t _func;
};
void Save(const string& message)
{
    const string target="./log.txt";
    FILE* fp=fopen(target.c_str(),"a+");
    if(!fp)
    {
        cerr<<" fopen error "<<endl;
        return;
    } 
    fputs(message.c_str(),fp);
    fputs("\n",fp);
    
    fclose(fp);
}

Block.hpp

#pragma once

#include<iostream>
#include<queue>
#include<pthread.h>
#include<unistd.h>

using namespace std;
static const int gmaxcap=5;
template<class T>
class BlockQueue
{
public:
    
    BlockQueue(const int& maxcap=gmaxcap)
        :_maxcap(maxcap)
    {
        pthread_mutex_init(&_mutex,nullptr);
        pthread_cond_init(&_pcond,nullptr);
        pthread_cond_init(&_ccond,nullptr);
    }
    void push(const T& in)//输入型参数，const &
    {
        pthread_mutex_lock(&_mutex);
        //1.判断
        while(is_full())//防止向队列中多push内容->提高代码健壮性
        {//生产条件不满足无法生产，需要生产者等待
        //临界区:
        //pthread_cond_wait   第二个参数必须为我们正在使用的互斥锁！！！
        //1.pthread_cond_wait:该函数在调用的时候，会以原子性的方式将锁释放，并将自己挂起
        //2.pthread_cond_wait:该函数在被唤醒返回的时候，会自动重新获取当时传入的锁
            pthread_cond_wait(&_pcond,&_mutex);
        }
        _q.push(in);
        //pthread_cond_signal即可在临界区内也可在临界区外
        pthread_cond_signal(&_ccond);
        pthread_mutex_unlock(&_mutex);

    }
    void pop(T* out)//输出型参数：*， // 输入输出型：&
    {
        pthread_mutex_lock(&_mutex);
        while(is_empty())
        {
            pthread_cond_wait(&_ccond,&_mutex);
        }
        *out=_q.front();
        _q.pop();
        pthread_cond_signal(&_pcond);
        pthread_mutex_unlock(&_mutex);
    }
    ~BlockQueue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_pcond);
        pthread_cond_destroy(&_ccond);
    }
private:
    bool is_empty() 
    {
        return _q.empty();
    }
    bool is_full()
    {
        return _q.size()==_maxcap;
    }
    queue<T> _q;
    int _maxcap;//队列中元素的上限
    pthread_mutex_t _mutex;
    pthread_cond_t _pcond;//生产者对应的条件变量
    pthread_cond_t _ccond;//消费者对应的条件变量
};

Main.cc

#include<time.h>
#include<cstring>
#include<sys/types.h>
#include"BlockQueue.hpp"
#include"Task.hpp"

const string oper="+-*/%";

template<class C,class S>
class BlockQueues
{
public:
    BlockQueue<C> *c_bq;
    BlockQueue<S> *s_bq;
};
void* productor(void *bqs_)
{
    //BlockQueue<Task>* bq=static_cast<BlockQueue<Task>*>(bq_);
    BlockQueue<CalTask>* bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->c_bq;

    while(true)
    {
        //生产活动-x先用随机数构建数据
        int x=rand()%10+1;
        int y=rand()%5;
        int operCode=rand()%oper.size();
        CalTask t(x,y,oper[operCode],mymath);
        bq->push(t);
        cout<<"productor thread,生产数据: "<<t.toTaskString()<<endl;
        sleep(1);
    }
    return nullptr;
}
void *consumer(void* bqs_)
{
    //BlockQueue<Task>* bq=static_cast<BlockQueue<Task>*>(bq_);
    BlockQueue<CalTask>* bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->c_bq;
    BlockQueue<SaveTask>* save_bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->s_bq;
    
    while(true)
    {
        //消费活动
        CalTask t;
        bq->pop(&t);
        string result=t();
        cout<<"消费任务: "<<result<<endl;
        
        // SaveTask save(result,Save);
        // save_bq->push(save);
        // cout<<"consumer thread,推送保存任务完成..."<<endl;

    }
    return nullptr;
}

void* saver(void* bqs_)
{
    BlockQueue<SaveTask>* save_bq= (static_cast<BlockQueues<CalTask,SaveTask>*>(bqs_))->s_bq;

    while(true)
    {
        SaveTask t;
        save_bq->pop(&t);
        t();
        cout<<"保存任务完成..."<<endl;
    }
}
int main()
{
    srand((unsigned long)time(nullptr));
    BlockQueues<CalTask,SaveTask> bqs;
    bqs.c_bq=new BlockQueue<CalTask>();
    bqs.s_bq=new BlockQueue<SaveTask>();

    pthread_t c,p,s;
    pthread_create(&c,nullptr,consumer,&bqs);
    pthread_create(&p,nullptr,productor,&bqs);
    // pthread_create(&s,nullptr,saver,&bqs);
    
    pthread_join(c,nullptr);
    pthread_join(p,nullptr);
    // pthread_join(s,nullptr);
    
    delete bqs.c_bq;
    delete bqs.s_bq;
    return 0;
}

POSIX信号量

POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。但POSIX可以用于线程间同步。

        pthread_mutex_lock(&_mutex);
        while(is_empty())
        {
            pthread_cond_wait(&_ccond,&_mutex);
        }
        *out=_q.front();
        _q.pop();
        pthread_cond_signal(&_pcond);
        pthread_mutex_unlock(&_mutex);