一个简单http_server的实现(4)-线程池

为什么要线程池

之前实现的那个简陋的服务器只能接受一个连接，这显然是不行的，所以我们可以用多线程来解决这个问题
但是如果来一个连接就开一个线程那么连接很多的时候就会有开很多的线程，还是会影响性能，所以需要线程池，即固定最大可以开的线程，然后将线程管理起来，来一个连接就把它加入工作队列，然后呢让空闲的线程去竞争这个任务。

线程池实现

类定义

#ifndef __THREAD_POOL__
#define __THREAD_POOL__
#include "http_server.h" 
namespace thp
{
    template <class T>
    class Thread_Pool
    {
        private:
            int threadnum;//线程池中线程的最大数量
            std::vector<std::thread>_thread;//线程池中的线程
            std::queue<T*>task_queue;//工作队列
            std::mutex _mtx;//互斥锁
            std::condition_variable cv;//条件变量
            std::atomic_bool is_runing;//判断线程池是否在运行中
            void work();//线程工作函数
        public:
            void stop();//关闭线程池
            void init();//初始化线程池
            void append(T*task);//往工作队列中添加任务,参数使用右值应用来减少拷贝
            explicit Thread_Pool(int num):threadnum(num)
            {
                init();
            }
            ~Thread_Pool()
            {
                stop();
            }
            Thread_Pool(const Thread_Pool&)=delete;
            Thread_Pool& operator =(const Thread_Pool& t)=delete;
    };
}
#endif

其中is_runing是原子变量，互斥锁和条件变量是来保证工作队列的访问的安全，同时该类是模板类，但是工作队列中的元素要保证是一个类，并且实现了一个run函数

初始化

template<class T>
void thp::Thread_Pool<T>::init()
{
    if(is_runing==false)
    {
        is_runing=true;
        for(int i=0;i<threadnum;i++)
        {
            _thread.emplace_back(std::thread(&thp::Thread_Pool<T>::work,this));//创建线程
        }
    }
}

创建threadnum个线程，线程的入口是本类的work函数

添加任务

template<class T>
void thp::Thread_Pool<T>::append(T*task)
{
    if(is_runing)
    {
        {
            std::unique_lock<std::mutex>lk(_mtx);//队列加锁
            task_queue.push(task);
        }
        cv.notify_one();//唤醒一个线程
    }
}

队列不为空就唤醒一个线程

线程的入口函数

template<class T>
void thp::Thread_Pool<T>::work()
{
    while(is_runing)
    {
        T* task=nullptr;
        {//函数块，保证uniquelock析构时解锁
            std::unique_lock<std::mutex> lk(_mtx);//给队列加锁
            if(!task_queue.empty())
            {
                task=task_queue.front();
                task_queue.pop();
            }else
                cv.wait(lk,[this]()->bool{return !is_runing||!task_queue.empty();});
        }
        if(task!=nullptr)
        {
            task->run();//调用任务类的run函数
            delete task;//执行完，释放空间
        }
    }
    return ;
}

关闭线程池

template<class T>
void thp::Thread_Pool<T>::stop()
{
    is_runing=false;
    cv.notify_all();//唤醒所有线程
    for(auto &t:_thread)//等待所有线程退出
    {
        if(t.joinable())
            t.join();
    }
    while(!task_queue.empty())//工作队列中还有
    {
        delete task_queue.front();
        task_queue.pop();
    }
}

测试

#include "../lib/thread_pool.cpp" 
class thp_test
{
    private:
        std::string s;
    public:
        thp_test():s("hello thread "){};
        void run()
        {
            std::cout<<s<<pthread_self()<<std::endl;
        }
};
int main()
{
    thp::Thread_Pool<thp_test>tp(10);
    sleep(1);
    for(int i=0;i<10;i++)
    {
        tp.append(new thp_test());
        sleep(1);
    }
    tp.stop();
    return 0;
}