【Linux】基于单例模式懒汉实现方式的线程池

清疚 2024-04-25 20:19 64阅读 0赞

## 基于单例模式懒汉实现方式的线程池 ##

### 一、LockGuard.hpp ###

#pragma once
    #include <iostream>
    #include <pthread.h>
    class Mutex//锁的对象
    {
        
    public:
        Mutex(pthread_mutex_t* lock_p=nullptr)
        :_lock_p(lock_p)
        {
        }
        ~Mutex()
        {
        }
        void lock()
        {
        
            if(_lock_p)
            {
        
                pthread_mutex_lock(_lock_p);
            }
        }
        void unlock()
        {
        
            if(_lock_p)
            {
        
                pthread_mutex_unlock(_lock_p);
            }
        }
    private:
        pthread_mutex_t* _lock_p;//锁的指针
    };
    class LockGuard//加锁和解锁的类
    {
        
    public:
        LockGuard(pthread_mutex_t* mutex)
        :_mutex(mutex)
        {
        
            _mutex.lock();//在构造函数进行加锁
        }
        ~LockGuard()
        {
        
            _mutex.unlock();//在析构函数进行解锁
        }
    private:
        Mutex _mutex;
    };

### 二、Task.hpp ###

#pragma once
    #include <iostream>
    #include <functional>
    #include <string>
    class Task
    {
        
        //using func=std::function<int(int,int,char)>;
        typedef std::function<int(int,int,char)> func_t;//函数对象
    public:
        Task()
        {
        }
        Task(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),"%d %c %d=%d",_x,_op,_y,result);//结果字符串
            return buffer;
        }
        std::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;//回调函数
    };
     
    const std::string oper = "+-*/%";
    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)
            {
        
                std::cerr << "div zero error" << std::endl;
                result = -1;
            }
            else
                result = x / y;
        }
            break;
        case '%':
        {
        
            if (y == 0)
            {
        
                std::cerr << "mod zero" << std::endl;
                result = -1;
            }
            else
                result = x % y;
        }
        break;
        default:
            std::cout<<"运算符输入有误"<<std::endl;
            break;
        }
        return result;
    }

### 三、Thread.hpp ###

#pragma once
    #include <iostream>
    #include <pthread.h>
    #include <string>
    #include <functional>
    #include <cassert>
    namespace ThreadNs
    {
        
        typedef std::function<void*(void*)> func_t;//定义一个函数对象类型，它的返回值和参数都是void*
        const int num = 1024;
     
        class Thread
        {
        
        private:
            //因为形参有个this指针，所以弄成static
            static void* start_routine(void* args)//这里的args就是 start()的ctx
            {
        
                Thread* _this=static_cast<Thread*>(args);
                return _this->callback();
            } 
            void* callback()
            {
        
                return _func(_args);
            }
        public:
            //构造函数
            Thread()
            {
        
                char nameBuffer[num];
                snprintf(nameBuffer,sizeof(nameBuffer),"thread-%d",threadNum++);
                _name=nameBuffer;
            }
            void start(func_t func,void* args=nullptr)//线程启动
            {
        
                _func=func;
                _args=args;
                int n=pthread_create(&_tid,nullptr,start_routine,this);
                assert(n==0);
                (void)n;
            }
            void join()
            {
        
                int n=pthread_join(_tid,nullptr);
                assert(n==0);
                (void)n;
            }
            std::string threadName()
            {
        
                return _name;
            }
            ~Thread()
            {
        }
        private:
            std::string _name;//线程的名字
            func_t _func;//线程的回调函数
            void* _args;//喂给线程的参数
            pthread_t _tid;//线程ID
            static int threadNum;//线程编号，最好自己再加个锁
        };
        int Thread::threadNum=1;
        //异常和if。意料之外
        //assert。意料之中。99%概率为真。
    }

### 四、ThreadPool.hpp ###

#pragma once
    #include <vector>
    #include <queue>
    #include <pthread.h>
    #include <unistd.h>
    #include <mutex>
    #include "Thread.hpp"
    #include "LockGuard.hpp"
    using namespace ThreadNs;
    const int gnum =5;
     
    template <class T>//声明
    class ThreadPool;
     
    template <class T>
    struct ThreadData
    {
        
        ThreadData(ThreadPool<T>* tp,const std::string& s)
        :_threadPool(tp)
        ,_name(s)
        {
        }
        ThreadPool<T>* _threadPool;
        std::string _name;
    };
    template <class T>
    class ThreadPool
    {
        
    private:
        //因为普通成员函数第一个参数是this指针，和回调方法不匹配，故改成static类型
        static void* handlerTask(void* args)//args是ThreadData对象指针
        {
        
            ThreadData<T>* td=static_cast<ThreadData<T>*>(args);
            while(1)
            {
        
                T t;
                {
           //RAII，出了作用域LockGuard会销毁，将析构锁
                    LockGuard lockGuard(td->_threadPool->mutex());//加锁
                    while(td->_threadPool->IsQueueEmpty())//如果队列为空，则等待
                    {
        
                        td->_threadPool->ThreadWait();
                    }
                    //线程能走到这里，说明队列一定有任务给线程
                    t=td->_threadPool->Pop();//从队列中取出任务
                }
                std::cout<<td->_name<<"已获取任务："<<t.toTaskString()<<"处理结果是："<<t()<<std::endl;//处理任务,这个任务放到线程的外面
            }
            delete td;//析构ThreadData对象
            return nullptr;
        }
        ThreadPool(const int& num=gnum)
        :_num(num)
        {
        
            pthread_mutex_init(&_mutex,nullptr);
            pthread_cond_init(&_cond,nullptr);
            //创建线程
            for(int i=0;i<_num;++i)
            {
        
                _threads.push_back(new Thread());
            }
        }
        ThreadPool(const ThreadPool<T>&)=delete;//禁用拷贝构造
        ThreadPool<T>& operator=(const ThreadPool<T>&)=delete;//禁用赋值运算符重载
     
    public://解决静态handlerTask是静态函数的问题，这几个都是偷家函数
        void LockQueue()   {
        pthread_mutex_lock(&_mutex);}
        void UnLockQueue() {
        pthread_mutex_unlock(&_mutex);}
        bool IsQueueEmpty(){
        return _taskQueue.empty();}
        void ThreadWait()  {
        pthread_cond_wait(&_cond,&_mutex);}
        T Pop()         
        {
        
            T t=_taskQueue.front();
            _taskQueue.pop();
            return t;
        } 
        pthread_mutex_t* mutex()
        {
        
            return &_mutex;
        }
    public: 
        void run()//线程启动
        {
        
            for(const auto& t:_threads)
            {
        
                ThreadData<T>* td=new ThreadData<T>(this,t->threadName());
                t->start(handlerTask,(void*)td);
                std::cout<<t->threadName()<<"start..."<<std::endl;
            }
        }
        void push(const T& in)
        {
        
            //RAII,出了作用域，锁将会被释放
            LockGuard lockGuard(&_mutex);
            _taskQueue.push(in);
            pthread_cond_signal(&_cond);
            std::cout<<"任务发送成功"<<std::endl;
        }
        ~ThreadPool()
        {
        
            pthread_mutex_destroy(&_mutex);
            pthread_cond_destroy(&_cond);
            for(const auto& t:_threads)
            {
        
                delete t;
            }
        }
        static ThreadPool<T>* getInstance()//这里的static的作用是让这个函数只有一份，获取单例对象。tp是临界资源，需要加锁
        {
        
            if(nullptr==tp)//因为锁只创建一次，防止线程进来被锁阻塞
            {
        
                //只进来一次就够了
                _singletonLock.lock();
                if(nullptr==tp)//说明对象还没有被创建
                {
        
                    tp=new ThreadPool<T>(); 
                }
                _singletonLock.unlock();
            }
            return tp;
        }
    private:
        int _num;//线程个数
        std::vector<Thread*> _threads;//使用vector存放线程
        std::queue<T> _taskQueue;//任务队列，往里面放任务，它是共享资源，需要加锁保护
        pthread_mutex_t _mutex;//互斥锁
        pthread_cond_t _cond;//条件变量
     
        static ThreadPool<T>* tp;//单例模式静态的对象指针
        static std::mutex _singletonLock;//获取单例对象使用的锁
     
    };
    template <class T>
    ThreadPool<T>* ThreadPool<T>::tp=nullptr;
     
    template <class T>
    std::mutex ThreadPool<T>::_singletonLock;