|
@@ -1,312 +0,0 @@
|
|
|
-#include "ThreadPool.h"
|
|
|
-
|
|
|
-#include "spdlog/spdlog.h"
|
|
|
-#include "fmt/std.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_threadMiniNum = 3;
|
|
|
- m_threadLiveNum = 0;
|
|
|
- m_threadExitNum = 0;
|
|
|
- m_threadAddNum = 5;
|
|
|
- m_threadMiniIdle = 2;
|
|
|
- m_threadMaxIdle = 6;
|
|
|
- m_threadRunNum = 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::milliseconds(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无法打印 */
|
|
|
- std::stringstream ss;
|
|
|
- ss << std::this_thread::get_id();
|
|
|
- SPDLOG_DEBUG("线程ID:{}退出任务循环", ss.str());
|
|
|
-
|
|
|
- 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);
|
|
|
-}
|