C++封装POSIX 线程库（六）线程池

逃离我推掉我的手 2022-06-15 04:58 226阅读 0赞

线程池有两个主要组件:

> 1.threads  
> 2.blocking queue

解决的问题是线程创建与销毁带来的开销和通过线程池实现伪异步  
过程类似于简单的生产者消费者问题(详见[wiki][])

## Blocking Queue ##

对应用户添加任务CallBack和线程拿任务执行两个操作,因此需要两个条件变量和一个互斥锁

如果把用户添加任务比做生产者,线程拿任务比做消费者..那么需要两个条件变量就解决了

> 1.生产者>消费者(防止”撑爆”任务队列)  
> 2.生产者<消费者(没有任务可执行时线程会挂起)

--------------------

## Threads ##

线程池由一组线程组成,构造阶段创建一组线程并挂起,每当添加一个任务的时候就notify一个线程执行…  
当线程池析构的时候,将这些线程回收….这里有个很大的坑…

--------------------

## 代码 ##

完整代码:[https://github.com/NearXdu/ThreadPool][https_github.com_NearXdu_ThreadPool]

#ifndef __THREADPOLL_H__
    #define __THREADPOLL_H__
    #include "MutexLock.h"
    #include "Condition.h"
    #include "Thread.h"
    #include <boost/noncopyable.hpp>
    #include <boost/function.hpp>
    #include <boost/bind.hpp>
    #include <boost/ptr_container/ptr_vector.hpp>
    #include <queue>
    class ThreadPool : boost::noncopyable
    {
        public:
        typedef boost::function<void()>Task;
        ThreadPool(size_t qsize,size_t nthreads);
        ~ThreadPool();
        void start();
        void stop();
        void addTask(Task t);
        Task getTask();
    
        bool isStarted()const{
       return isStarted_;}
        void runInThread();
    
        private:
        mutable MutexLock mutex_;
        Condition notEmpty_;//blocking queue  full条件变量
        Condition notFull_;//blocking queue empty 条件变量
        size_t queueSize_;//任务队列大小
        std::queue<Task> queue_;//任务队列
    
        size_t threadsNum_;//线程数目
        boost::ptr_vector<Thread> threads_;//线程池中的线程
        volatile bool isStarted_;
    
    };
    #endif
    
    
    ///
    #include "ThreadPool.h"
    ThreadPool::ThreadPool(size_t q,size_t n)
        :mutex_(),
         notEmpty_(mutex_),
         notFull_(mutex_),
         queueSize_(q),
         threadsNum_(n),
         isStarted_(false)
    {}
    ThreadPool::~ThreadPool()
    {
        if(isStarted_)
        {
        stop();
        }
    }
    
    
    void ThreadPool::stop()
    {   
        {
        MutexLockGuard lock(mutex_);
        isStarted_ = false;
        notEmpty_.notifyAll(); //激活所有的线程
    
        }
        for_each(threads_.begin(),
            threads_.end(),
            boost::bind(&Thread::join, _1));
    
    }
    void ThreadPool::addTask(Task t)
    {
        if(threads_.empty())
        {
        t();
        }
        else
        {
    
        MutexLockGuard lock(mutex_);
        while(queue_.size()>=queueSize_&&isStarted_)
            notFull_.wait();
        if(!isStarted_)
            return ;
    
        queue_.push(std::move(t));
        notEmpty_.notify();
        }
    }
    
    ThreadPool::Task ThreadPool::getTask()
    {
        MutexLockGuard lock(mutex_);
        while(queue_.empty()&&isStarted_)
        {
        notEmpty_.wait();
        }
        Task t;
        if(!queue_.empty())
        {
        assert(!queue_.empty());
        t=queue_.front();
        queue_.pop();
        notFull_.notify();
        }
    
        return t;
    }
    
    void ThreadPool::runInThread()
    {
        while(isStarted_)
        {
        //  std::cout<<"hello runinThread"<<std::endl;
        Task t(getTask());//will blocked until "add" ops
        //assert(t!=NULL);
        if(t)
        {
            t();//run
        }
        }
    }
    
    void ThreadPool::start()
    {
        isStarted_=true;
        threads_.reserve(threadsNum_);
        for(size_t i=0;i<threadsNum_;++i)
        {
        threads_.push_back(
            new Thread(boost::bind(
                &ThreadPool::runInThread,this)));
    
        threads_[i].start();
        }
    
    }

[wiki]: https://en.wikipedia.org/wiki/Thread_pool
[https_github.com_NearXdu_ThreadPool]: https://github.com/NearXdu/ThreadPool