External/module/ThreadPool/ThreadPool_fmtlog.cpp_

#include "ThreadPool.h"

#include "FmtLog/fmtlog.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_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);

    FMTLOG_DEBUG("***** Hello ThreadPool *****");

    // /* 创建初始的numThread个线程 */
    // createThread(numThread);
}
/* 析构函数 */
ThreadPool::~ThreadPool()
{
    FMTLOG_INFO("线程池正在退出...");
    /* 将stop置为true */
    {
        std::unique_lock<std::mutex> lock(m_mutexTask);
        m_stop = true;
    }

    FMTLOG_INFO("通知所有工作线程退出...");
    /* 发送条件变量，通知所有线程 */
    m_cond_Task.notify_all();
    /* 等待所有的线程退出并回收完成 */
    while (!m_mapThreads.empty())
    {
        /* 管理线程自己退出了，所以要手动清空容器 */
        clearThread();
        std::this_thread::sleep_for(std::chrono::microseconds(100));
    }

    FMTLOG_INFO("回收管理线程...");
    /* 先回收管理线程 */
    m_managerThread.join();
    FMTLOG_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();
    FMTLOG_DEBUG("线程ID:{}退出任务循环", ss.str());

    return;
}


/**
 * @brief 管理者线程，维护线程创建或死亡
 * 
 */
void ThreadPool::managerThread()
{
    while (!m_stop)
    {
        /* 获取空闲线程的个数 */
        int num_idle = m_threadLiveNum.load() - m_threadRunNum.load();
        /* 判断线程是否够用，是否需要创建新的线程 */
        // FMTLOG_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)
            {
                // FMTLOG_INFO("需要添加{}个线程", numAdd);
                createThread(numAdd);
                continue;   /* 直接下一个循环，无需检查需不需要销毁线程 */
            }
            
        }

        /* 判断空闲线程是否过多，是否需要销毁几个线程 */
        // FMTLOG_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);

            FMTLOG_INFO("有{}个线程需要退出", m_threadExitNum.load());
            /* 唤醒需要退出的num_idle个线程 */
            for (int i = 0; i < num_Exit; i++)
            {
                m_cond_Task.notify_one();
            }
        }
        /* 回收退出的线程 */
        clearThread();

        // FMTLOG_INFO("线程池中的线程实例个数:{}", m_threads.size());
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    FMTLOG_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);
}