|  | @@ -0,0 +1,316 @@
 | 
	
		
			
				|  |  | +#include "ThreadPool.h"
 | 
	
		
			
				|  |  | +#include "spdlog/spdlog.h"
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/**
 | 
	
		
			
				|  |  | + * @brief Construct a new Thread Pool:: Thread Pool object
 | 
	
		
			
				|  |  | + *        构造函数,从这里创建线程,和Linux C的线程池不同,Linux C的线程池数组管理的是线程ID,
 | 
	
		
			
				|  |  | + *        而C++的线程数组直接存储的就是线程函数体,里面有个function的变量,指向任务队列中任务函数
 | 
	
		
			
				|  |  | + * 
 | 
	
		
			
				|  |  | + * @param numThread 
 | 
	
		
			
				|  |  | + */
 | 
	
		
			
				|  |  | +ThreadPool::ThreadPool() : 
 | 
	
		
			
				|  |  | +    m_stop(false)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    /* 初始化变量 */
 | 
	
		
			
				|  |  | +    // m_threadMaxNum = std::thread::hardware_concurrency();   /* 根据CPU核心数规定线程数目 */
 | 
	
		
			
				|  |  | +    m_threadMaxNum = 256;
 | 
	
		
			
				|  |  | +    m_threadMiniNum = 2;
 | 
	
		
			
				|  |  | +    m_threadAddNum = 2;
 | 
	
		
			
				|  |  | +    m_threadMiniIdle = 1;
 | 
	
		
			
				|  |  | +    m_threadMaxIdle = 4;
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    m_threadRunNum = 0;
 | 
	
		
			
				|  |  | +    m_threadLiveNum = 0;
 | 
	
		
			
				|  |  | +    m_threadExitNum = 0;
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    /* 创建管理线程,this表示是这个类的成员函数 */
 | 
	
		
			
				|  |  | +    m_managerThread = std::thread(&ThreadPool::managerThread, this);
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    SPDLOG_DEBUG("***** Hello ThreadPool *****");
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    // /* 创建初始的numThread个线程 */
 | 
	
		
			
				|  |  | +    // createThread(numThread);
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +/* 析构函数 */
 | 
	
		
			
				|  |  | +ThreadPool::~ThreadPool()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    SPDLOG_INFO("线程池正在退出...");
 | 
	
		
			
				|  |  | +    /* 将stop置为true */
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        std::unique_lock<std::mutex> lock(m_mutexTask);
 | 
	
		
			
				|  |  | +        m_stop = true;
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    SPDLOG_INFO("通知所有工作线程退出...");
 | 
	
		
			
				|  |  | +    /* 发送条件变量,通知所有线程 */
 | 
	
		
			
				|  |  | +    m_cond_Task.notify_all();
 | 
	
		
			
				|  |  | +    /* 等待所有的线程退出并回收完成 */
 | 
	
		
			
				|  |  | +    while (!m_mapThreads.empty())
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        /* 管理线程自己退出了,所以要手动清空容器 */
 | 
	
		
			
				|  |  | +        clearThread();
 | 
	
		
			
				|  |  | +        std::this_thread::sleep_for(std::chrono::microseconds(100));
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    SPDLOG_INFO("回收管理线程...");
 | 
	
		
			
				|  |  | +    /* 先回收管理线程 */
 | 
	
		
			
				|  |  | +    m_managerThread.join();
 | 
	
		
			
				|  |  | +    SPDLOG_INFO("===== 线程池退出完成 =====");
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 工作线程函数 */
 | 
	
		
			
				|  |  | +void ThreadPool::worker()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    m_threadLiveNum++;
 | 
	
		
			
				|  |  | +    while (true)
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        /* 等待任务队列中有任务 */
 | 
	
		
			
				|  |  | +        std::unique_lock<std::mutex> lock(m_mutexTask);
 | 
	
		
			
				|  |  | +        /* 这里的wait第二个参数是lamba表达式,被唤醒后再次检查条件是否满足 */
 | 
	
		
			
				|  |  | +        m_cond_Task.wait(lock, [this]
 | 
	
		
			
				|  |  | +                               { return !m_queue_Tasks.empty() || m_stop ||
 | 
	
		
			
				|  |  | +                                        (m_threadExitNum > 0); });
 | 
	
		
			
				|  |  | +        /* 任务队列中有任务了,条件变量被唤醒了,先判断是不是需要结束线程 */
 | 
	
		
			
				|  |  | +        if (m_stop && m_queue_Tasks.empty())
 | 
	
		
			
				|  |  | +        {
 | 
	
		
			
				|  |  | +            break;
 | 
	
		
			
				|  |  | +        }
 | 
	
		
			
				|  |  | +        /* 判断是不是需要销毁多余的线程 */
 | 
	
		
			
				|  |  | +        if (m_threadExitNum.load() > 0 )
 | 
	
		
			
				|  |  | +        {
 | 
	
		
			
				|  |  | +            m_threadExitNum--;
 | 
	
		
			
				|  |  | +            /* 再次判断有没有新任务,有就不退出 */
 | 
	
		
			
				|  |  | +            if ( m_queue_Tasks.empty())
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                break;
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +        }
 | 
	
		
			
				|  |  | +        /* 取出任务,执行任务 */
 | 
	
		
			
				|  |  | +        std::function<void()> task(std::move(m_queue_Tasks.front()));
 | 
	
		
			
				|  |  | +        m_queue_Tasks.pop();  /* 取出的任务出队 */
 | 
	
		
			
				|  |  | +        lock.unlock();              /* 解锁任务队列 */
 | 
	
		
			
				|  |  | +        m_threadRunNum++;     /* 更新线程状态数 */
 | 
	
		
			
				|  |  | +        /* 开始执行任务 */
 | 
	
		
			
				|  |  | +        task();
 | 
	
		
			
				|  |  | +        m_threadRunNum--;         /* 更新线程状态数 */
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    /* 线程结束 */
 | 
	
		
			
				|  |  | +    m_threadLiveNum--;
 | 
	
		
			
				|  |  | +    /* 将自身ID加入到退出列表中 */
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        std::unique_lock<std::mutex> lock(m_mutexExitThreadID);
 | 
	
		
			
				|  |  | +        m_exitThreadID.emplace_back(std::this_thread::get_id());
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +    /* 使用流将线程ID转换成字符串,不然fmt无法打印
 | 
	
		
			
				|  |  | +     * 这里通过hash转换成了size_t */
 | 
	
		
			
				|  |  | +    // std::stringstream ss;
 | 
	
		
			
				|  |  | +    // ss << std::this_thread::get_id();
 | 
	
		
			
				|  |  | +    auto threadID = std::this_thread::get_id();
 | 
	
		
			
				|  |  | +    std::hash<std::thread::id> hasher;
 | 
	
		
			
				|  |  | +    size_t id = hasher(threadID);
 | 
	
		
			
				|  |  | +    SPDLOG_DEBUG("线程ID:{}退出任务循环", id);
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    return;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/**
 | 
	
		
			
				|  |  | + * @brief 管理者线程,维护线程创建或死亡
 | 
	
		
			
				|  |  | + * 
 | 
	
		
			
				|  |  | + */
 | 
	
		
			
				|  |  | +void ThreadPool::managerThread()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    while (!m_stop)
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        /* 获取空闲线程的个数 */
 | 
	
		
			
				|  |  | +        int num_idle = m_threadLiveNum.load() - m_threadRunNum.load();
 | 
	
		
			
				|  |  | +        /* 判断线程是否够用,是否需要创建新的线程 */
 | 
	
		
			
				|  |  | +        // SPDLOG_DEBUG("***** 判断是否需要添加线程 *****");
 | 
	
		
			
				|  |  | +        if ((num_idle < m_threadMiniIdle.load()) && (m_threadLiveNum.load() < m_threadMaxNum) )
 | 
	
		
			
				|  |  | +        {
 | 
	
		
			
				|  |  | +            std::unique_lock<std::mutex> lock(m_mutexTask);
 | 
	
		
			
				|  |  | +            int numTask = (int)m_queue_Tasks.size();            /* 获取任务队列中的任务个数 */
 | 
	
		
			
				|  |  | +            lock.unlock();
 | 
	
		
			
				|  |  | +            int numAdd = 0;
 | 
	
		
			
				|  |  | +            if(numTask > 0)
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                /* 任务数 + 存在的线程个数是否大于最大线程数 */
 | 
	
		
			
				|  |  | +                if( numTask + m_threadLiveNum.load() <= m_threadMaxNum )
 | 
	
		
			
				|  |  | +                {
 | 
	
		
			
				|  |  | +                    /* 创建numTask个线程 */
 | 
	
		
			
				|  |  | +                    numAdd = numTask;
 | 
	
		
			
				|  |  | +                }
 | 
	
		
			
				|  |  | +                /* 默认添加的个数 + 存在的线程数是否大于最大线程数 */
 | 
	
		
			
				|  |  | +                else if ( (m_threadAddNum.load() + m_threadLiveNum.load()) <= m_threadMaxNum)
 | 
	
		
			
				|  |  | +                {
 | 
	
		
			
				|  |  | +                    /* 创建m_threadAddNum个线程 */
 | 
	
		
			
				|  |  | +                    numAdd = m_threadAddNum.load();
 | 
	
		
			
				|  |  | +                }
 | 
	
		
			
				|  |  | +                /* 能添加几个线程就添加几个线程 */
 | 
	
		
			
				|  |  | +                else
 | 
	
		
			
				|  |  | +                {
 | 
	
		
			
				|  |  | +                    numAdd = m_threadMaxNum - m_threadLiveNum.load();
 | 
	
		
			
				|  |  | +                }
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +            /* 空闲线程数低于设置的最小空闲线程数 */
 | 
	
		
			
				|  |  | +            else
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                numAdd = m_threadMiniIdle.load() - num_idle;
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +            
 | 
	
		
			
				|  |  | +            if(numAdd > 0)
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                // SPDLOG_INFO("需要添加{}个线程", numAdd);
 | 
	
		
			
				|  |  | +                createThread(numAdd);
 | 
	
		
			
				|  |  | +                continue;   /* 直接下一个循环,无需检查需不需要销毁线程 */
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +            
 | 
	
		
			
				|  |  | +        }
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +        /* 判断空闲线程是否过多,是否需要销毁几个线程 */
 | 
	
		
			
				|  |  | +        // SPDLOG_DEBUG("***** 判断是否需要销毁线程 *****");
 | 
	
		
			
				|  |  | +        /* 由于没规定每次销毁的线程个数,所以这里使用m_threadAddNum作为每次销毁的标准个数 */
 | 
	
		
			
				|  |  | +        if (num_idle > m_threadMaxIdle.load())
 | 
	
		
			
				|  |  | +        {
 | 
	
		
			
				|  |  | +            int num_Exit = num_idle = m_threadMaxIdle.load();
 | 
	
		
			
				|  |  | +            if (num_Exit > m_threadAddNum.load())
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                num_Exit = m_threadAddNum.load();
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +            m_threadExitNum.exchange(num_Exit);
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +            SPDLOG_INFO("有{}个线程需要退出", m_threadExitNum.load());
 | 
	
		
			
				|  |  | +            /* 唤醒需要退出的num_idle个线程 */
 | 
	
		
			
				|  |  | +            for (int i = 0; i < num_Exit; i++)
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                m_cond_Task.notify_one();
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +        }
 | 
	
		
			
				|  |  | +        /* 回收退出的线程 */
 | 
	
		
			
				|  |  | +        clearThread();
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +        // SPDLOG_INFO("线程池中的线程实例个数:{}", m_threads.size());
 | 
	
		
			
				|  |  | +        std::this_thread::sleep_for(std::chrono::seconds(1));
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +    SPDLOG_INFO("管理线程退出...");
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/**
 | 
	
		
			
				|  |  | + * @brief 创建新的线程
 | 
	
		
			
				|  |  | + *        注意:这里只能使用lambda表达式,或者将do_work变成全局函数,emplace_back会调用thread构造函数将lambda表达式构造成一个std::thread实例
 | 
	
		
			
				|  |  | + *        lambda表达式里是子线程,外面是主线程
 | 
	
		
			
				|  |  | + * 
 | 
	
		
			
				|  |  | + */
 | 
	
		
			
				|  |  | +void ThreadPool::createThread(int num)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    for (int i = 0; i < num; i++)
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        /* 创建线程,传入工作函数 */
 | 
	
		
			
				|  |  | +        std::thread t(&ThreadPool::worker, this);
 | 
	
		
			
				|  |  | +        m_mapThreads.insert(std::make_pair( t.get_id(), std::move(t) ));
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/**
 | 
	
		
			
				|  |  | + * @brief 删除线程池中失效的线程实例,使用递归的方法遍历全部
 | 
	
		
			
				|  |  | + * 
 | 
	
		
			
				|  |  | + */
 | 
	
		
			
				|  |  | +void ThreadPool::clearThread()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    for(auto& it : m_exitThreadID)
 | 
	
		
			
				|  |  | +    {
 | 
	
		
			
				|  |  | +        auto it1 = m_mapThreads.find(it);
 | 
	
		
			
				|  |  | +        if(it1 != m_mapThreads.end())
 | 
	
		
			
				|  |  | +        {
 | 
	
		
			
				|  |  | +            if(it1->second.joinable())
 | 
	
		
			
				|  |  | +            {
 | 
	
		
			
				|  |  | +                it1->second.join();
 | 
	
		
			
				|  |  | +                m_mapThreads.erase(it1);
 | 
	
		
			
				|  |  | +            }
 | 
	
		
			
				|  |  | +        }
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +    m_exitThreadID.clear();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 获取线程池最大线程的个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadMaxNum()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadMaxNum;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 设置线程池最大线程的个数 */
 | 
	
		
			
				|  |  | +void ThreadPool::setThreadMaxNum(int num)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    m_threadMaxNum = num;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 获取线程池最大线程的个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadMiniNum()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadMiniNum;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 设置线程池最大线程的个数 */
 | 
	
		
			
				|  |  | +void ThreadPool::setThreadMiniNum(int num)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    m_threadMiniNum = num;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 获取线程池空闲线程的个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadIdleNum()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadLiveNum.load() - m_threadRunNum.load();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 获取线程池正在运行的线程个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadRunNum()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadRunNum.load();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | + /* 获取线程池现存的线程个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadLiveNum()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadLiveNum.load();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 线程池每次创建线程的个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadAddNum()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadAddNum.load();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 设置线程池每次创建线程的个数 */
 | 
	
		
			
				|  |  | +void ThreadPool::setThreadAddNum(int num)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    m_threadAddNum.exchange(num);
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 线程池最小空闲线程的个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadMiniIdle()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadMiniIdle.load();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 设置线程池最小空闲线程的个数 */
 | 
	
		
			
				|  |  | +void ThreadPool::setThreadMiniIdle(int num)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    m_threadMiniIdle.exchange(num);
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 线程池最大空闲线程的个数 */
 | 
	
		
			
				|  |  | +int ThreadPool::getThreadMaxIdle()
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    return m_threadMaxIdle.load();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +/* 设置线程池最大空闲线程的个数 */
 | 
	
		
			
				|  |  | +void ThreadPool::setThreadMaxIdle(int num)
 | 
	
		
			
				|  |  | +{
 | 
	
		
			
				|  |  | +    m_threadMaxIdle.exchange(num);
 | 
	
		
			
				|  |  | +}
 |