Linux下任务队列类

秒速五厘米 2022-08-18 05:52 188阅读 0赞

## 任务队列主要满足以下需求： ##

*  可定制：可个性化任务队列的名称，最大任务数和任务处理函数
 *  多个任务队列并存，通过任务队列名称唯一标记。类似rabbitMQ
 *  每个任务支持二进制安全，可存储各种结构的信息（二进制结构体&字符串….）
 *  支持多线程操作，支持多线程同步
 *  支持单例模式

## talk is cheap, show me the code !!! ##

### taskQueue.h ###

#ifndef __TASK_QUEUE_H
    #define __TASK_QUEUE_H
    
    #include <iostream>
    #include <string>
    #include <queue>
    #include <map>
    #include <pthread.h>
    
    //默认任务包大小为1k
    #define MAX_TASK_LEN 1024
    
    using namespace std;
    
    class Task
    {
    public:
        //任务内容，通过taskContentLen判断长度，二进制安全 
        char taskContent[MAX_TASK_LEN];
        unsigned int taskContentLen;
    
        Task(){
            memset(taskContent,0,sizeof(taskContent));
            taskContentLen= 0;
        }
        //任务处理函数
        bool doWith(){
            cout<<"CLASS Task: "<<  taskContent<< endl;
        };
    };
    
    
    struct taskInfo
    {
        queue<Task *> *tasks;   //任务队列，任务空间可以使用其他空间配置器的方法
        pthread_mutex_t lock;   //任务队列互斥锁,控制多线程时tasks中入队&出队的同步问题
        pthread_cond_t cond;    //任务队列条件变量,控制最大任务数
        unsigned int taskNum;   //任务队列中任务数
        unsigned int maxTaskNum;    //任务队列最大任务数
        taskInfo(unsigned int maxTaskNums):maxTaskNum(maxTaskNums){
            taskNum= 0;
            pthread_mutex_init(&lock, NULL);
        }
    };
    
    class TaskQueue
    {
    private:
        typedef map<string,taskInfo *> map_task;
        typedef map<string,taskInfo *>::iterator map_task_iter;
    
        pthread_mutex_t taskQueuesLock; //多线程注册taskQueues时同步问题
        map_task taskQueues;    //可以命名多个任务队列
        TaskQueue();
    
    public:
        bool PushTask(string taskName,Task* one);           //增加任务至任务队列
        Task *PopTask(string taskName);                     //从任务队列中取出任务
    
        bool RegisterQueue(string name,unsigned int maxTaskNum);    //注册固定名称的任务队列并初始化任务队列中的最大任务数
    
        //任务空间分配器可以使用其他中间件或者库完成，所以设置成virtual
        virtual Task* MallocTask(); 
        virtual void FreeTask(Task *pTask);
        virtual ~TaskQueue();
        //单例模式
        static TaskQueue & GetInstance(){
            static TaskQueue instance;
            return instance;
        }
    };
    
    #endif

### taskQueue.cpp ###

#include "taskQueue.h"
    
    /*
    *@fun:  初始化锁
    */
    TaskQueue::TaskQueue()
    {
        pthread_mutex_init(&taskQueuesLock, NULL);
    }
    
    /*
    *@fun:  释放所有空间和锁
    */
    TaskQueue::~TaskQueue()
    {
        Task *pTask= NULL;
        for(map_task_iter it= taskQueues.begin(); it!= taskQueues.end();it++)
        {
            if(it->second!= NULL)
            {
                //释放任务队列中申请空间
                while(!it->second->tasks->empty())
                {
                    pTask= it->second->tasks->front();
                    it->second->tasks->pop();
                    FreeTask(pTask);
                }
                delete it->second->tasks;
                pthread_mutex_destroy(&(it->second->lock));
                pthread_cond_destroy(&(it->second->cond));
                free(it->second);
                it->second= NULL;
            }
        }
    }
    
    /*
    *@fun:          注册任务队列
    *@param1:       任务队列名称
    *@param2:       设置任务队列的最大任务数
    *@ret:          true/注册成功   false/注册失败
    */
    bool TaskQueue::RegisterQueue(string name,unsigned int maxTaskNum)
    {
        bool bRet= false;
    
        if(name.size()> 0)
        {
            pthread_mutex_lock(&taskQueuesLock);
            if(taskQueues.find(name)== taskQueues.end())
            {
                taskInfo *pTest= (taskInfo *)malloc(sizeof(taskInfo));
                if(pTest!= NULL)
                {
                    //申请队列空间
                    pTest->tasks= new queue<Task *>;
                    if(pTest->tasks!= NULL)
                    {
                        //初始化taskInfo
                        pTest->taskNum= 0;
                        pTest->maxTaskNum= maxTaskNum;
                        pthread_mutex_init(&(pTest->lock), NULL);
                        pthread_cond_init(&(pTest->cond),NULL);
                        //此处insert函数使用位置的思考
                        taskQueues.insert(make_pair(name,pTest));
                        bRet= true;
                    }
                    else
                    {
                        free(pTest);    
                    }
                }
            }
            pthread_mutex_unlock(&taskQueuesLock);
        }
        return bRet;
    }
    
    /*
    *@fun:          释放任务节点空间
    */
    void TaskQueue::FreeTask(Task *pTask)
    {
        if(pTask!= NULL)
        {
            delete pTask;
        }
    }
    
    /*
    *@fun:          申请任务节点空间
    *@ret:          NULL/申请失败   非NULL/申请成功
    */
    Task* TaskQueue::MallocTask()
    {
        Task *pTask=  new Task;
        return pTask;
    }
    
    /*
    *@fun:          添加任务至任务队列
    *@param1:       任务队列名称
    *@param2:       任务节点指针
    *@ret:          true/添加成功   false/添加失败
    */
    bool TaskQueue::PushTask(string taskName,Task* one)
    {
        bool bRet= false;
        map_task_iter it= taskQueues.find(taskName);
    
        if(it!= taskQueues.end())
        {
            //加入互斥锁可解决多线程之间的同步问题
            pthread_mutex_lock(&(it->second->lock));
    
            //如果任务队列中任务数已经到达最大上限时，等待
            if(it->second->taskNum>= it->second->maxTaskNum)
            {
                pthread_cond_wait(&(it->second->cond),&(it->second->lock));
            }
            else
            {
                it->second->tasks->push(one);
                it->second->taskNum++;
            }
            pthread_mutex_unlock(&(it->second->lock));
            bRet= true;
        }
        return bRet;
    }
    
    /*
    *@fun:          从任务队列中取出任务
    *@param1:       任务队列名称
    *@ret:          NULL/取出失败   非NULL/取出成功
    */
    Task *TaskQueue::PopTask(string taskName)
    {
        Task* pTask= NULL;
        map_task_iter it= taskQueues.find(taskName);
    
        if(it!= taskQueues.end() && it->second->tasks!=NULL && !it->second->tasks->empty())
        {
            //加入互斥锁可解决多线程之间的同步问题
            pthread_mutex_lock(&(it->second->lock));
    
            //如果任务队列中任务数小于最大任务数，发送信号
            if(it->second->taskNum<= it->second->maxTaskNum)
            {
                pTask= it->second->tasks->front();
                it->second->tasks->pop();
                it->second->taskNum--;
                pthread_cond_signal(&(it->second->cond));
            }
    
            pthread_mutex_unlock(&(it->second->lock));
        }
        return pTask;
    }

### 示例函数：main.cpp ###

#include <iostream>
    #include "taskQueue.h"
    
    //PrintTask & HelloTask & WorldTask分别重载不同的任务处理函数 doWith
    class PrintTask: public Task
    {
    public:
        bool doWith(){
            std::cout<< "PrintTask: "<< taskContent<< std::endl;
        };
    };
    
    class HelloTask: public Task
    {
    public:
        bool doWith(){
            std::cout<< "HelloTask: "<< taskContent<< std::endl;
        };
    };
    
    class WorldTask: public Task
    {
    public:
        bool doWith(){
            std::cout<< "WorldTask: "<< taskContent<< std::endl;
        };
    };
    
    
    using namespace std;
    int main()
    {
        int num= 50;
        TaskQueue tq= TaskQueue::GetInstance();
        Task *pptask= NULL;
    
    // PrintTask *pt= NULL;
    // HelloTask *pt= NULL;
        WorldTask *pt= NULL;
    
        //注册名称为test的队列
        tq.RegisterQueue("test",100);
    
        for(int i= 0;i< num;i++)
        {
            pptask= tq.MallocTask();
    
            if(pptask!= NULL)
            {
                sprintf(pptask->taskContent," %dth task ->do something !",i);
                pptask->taskContentLen= 20;
                if(!tq.PushTask("test",pptask))
                    break;
            }
        }
    
        for(int i= 0;i< num;i++)
        {
        // pt= (PrintTask *)tq.PopTask("test");
        // pt= (HelloTask *)tq.PopTask("test");
            pt= (WorldTask *)tq.PopTask("test");
            pt->doWith();
            tq.FreeTask(pt);
            pt= NULL;
        }
        return 1;
    }

### 以上代码仅为个人愚见，如有疑问请各位看官指出。tks ###