C++缓存线程池CachedThreadPool原理、实现与对比解析

前言

在高并发编程的实战中，线程池算得上是优化性能、降低资源消耗的“老熟人”了。不过，相比那些线程数量一成不变的固定线程池，缓存线程池（CachedThreadPool） 显得更“聪明”一些——它能根据手头任务的多寡，动态调整线程数量：任务扎堆的时候，就多创建几个线程分摊压力；空闲下来时，又把多余的线程回收掉，既不浪费资源，又撑得住高并发。

一、缓存线程池的核心设计理念

说到底，缓存线程池要解决的核心问题是：任务量忽高忽低的时候，线程资源怎么管才高效？它的设计围绕着下面四个目标展开：

• 动态扩缩容：核心线程数量保持稳定，任务积压时先临时创建新线程顶上，等空闲线程超时了再自动回收；
• 任务异步调度：任务提交和执行之间通过一个线程安全的任务队列来解耦，谁也别打扰谁；
• 资源可控：最大线程数得有个上限，免得线程无限制地创建下去，把系统资源耗光；
• 优雅退出：线程池关闭的时候，得等着所有手头的任务执行完毕再停，不能丢了任务。

具体到代码实现上，这块缓存线程池的核心结构主要包含：

• 一个线程安全的任务队列（SyncQueue），专门存待执行的任务，支持阻塞或超时获取；
• 线程管理模块，负责维护线程集合，统计核心线程、空闲线程和当前总线程数；
• 动态扩缩容逻辑，根据空闲线程数和任务队列的状态来决定是创建还是回收线程；
• 任务提交接口，既能丢普通任务，也能提交带返回值的异步任务。

二、核心组件拆解

2.1 线程安全的任务队列（SyncQueue）

任务队列就像是线程池的“消息中枢”，它必须保证在多线程环境下读写安全，除此之外还得具备几个硬核能力：队列满了的时候，提交者得阻塞等待；队列空的时候，获取者也得耐心等着；并且支持超时退出。

核心实现细节

#include 
#include 
#include 
#include 
#include 
#include 
#include 
using namespace std;
#include "Logger.hpp"
#define QUESTOP 2
#define QUEEMPTY 1
#define ok 0
#ifndef SYNC_QUEUE_1_HPP
#define SYNC_QUEUE_1_HPP
namespace tulun
{
    static const size_t MaxTaskCount = 500;
    template 
    class SyncQueue
    {
    private:
        std::deque m_queue;
        mutable std::mutex m_mutex;
        std::condition_variable m_notEmpty;
        std::condition_variable m_notFull;
        std::condition_variable m_waitStop;
        int m_maxSize;
        bool m_needStop;
        int m_waitTime = 100;
        bool IsFull() const
        {
            bool full = m_queue.size() >= m_maxSize;
            return full;
        }
        bool IsEmpty() const
        {
            bool empty = m_queue.empty();
            return empty;
        }
        template 
        int Add(F &&task)
        {
            std::unique_lock locker(m_mutex);
            auto tag = m_notFull.wait_for(locker,
                                          std::chrono::milliseconds(m_waitTime),
                                          [this]() -> bool
                                          {
                                              return m_needStop || !IsFull();
                                          });
            if (m_needStop)
            {
                return 2;
            }
            if (!tag)
            {
                return 1;
            }
            m_queue.push_back(std::forward(task));
            m_notEmpty.notify_all();
            return 0;
        }
    public:
        SyncQueue(int maxsize = MaxTaskCount)
            : m_maxSize(maxsize),
              m_needStop(false)
        {
        }
        ~SyncQueue()
        {
            if (!m_needStop)
            {
                Stop();
            }
        }
        SyncQueue(const SyncQueue &) = delete;
        SyncQueue &operator=(const SyncQueue &) = delete;
        int Put(const T &task)
        {
            return Add(task);
        }
        int Put(T &&task)
        {
            return Add(std::forward(task));
        }
        int Take(T &task)
        {
            std::unique_lock locker(m_mutex);
            bool tag = m_notEmpty.wait_for(locker,
                                           std::chrono::milliseconds(m_waitTime),
                                           [this]() -> bool
                                           {
                                               return m_needStop || !IsEmpty();
                                           });
            if (m_needStop)
            {
                return 2;
            }
            if (!tag)
            {
                return 1;
            }
            task = m_queue.front();
            m_queue.pop_front();
            m_notFull.notify_all();
            return 0;
        }
        int Task(std::deque &tqu)
        {
            std::unique_lock locker(m_mutex);
            bool tag = m_notEmpty.wait_for(locker,
                                           std::chrono::milliseconds(m_waitTime),
                                           [this]() -> bool
                                           {
                                               return m_needStop || !IsEmpty();
                                           });
            if (m_needStop)
            {
                return 2;
            }
            if (!tag)
            {
                return 1;
            }
            tqu = std::move(m_queue);
            m_notFull.notify_all();
            return 0;
        }
        void Stop()
        {
            {
                std::unique_lock locker(m_mutex);
                m_needStop = true;
            }
            m_notEmpty.notify_all();
            m_notFull.notify_all();
        }
        void WaitQueueEmptyStop()
        {
            std::unique_lock locker(m_mutex);
            while (!IsEmpty())
            {
                m_waitStop.wait_for(locker, std::chrono::milliseconds(m_waitTime));
            }
            m_needStop = true;
            m_notFull.notify_all();
            m_notEmpty.notify_all();
        }
        bool Empty() const
        {
            std::unique_lock locker(m_mutex);
            return m_queue.empty();
        }
        bool Full() const
        {
            std::unique_lock locker(m_mutex);
            return m_queue.size() >= m_maxSize;
        }
        size_t Size() const
        {
            std::unique_lock locker(m_mutex);
            return m_queue.size();
        }
        size_t Count() const
        {
            return m_queue.size();
        }
    };
}
#endif

关键设计点

• 双条件变量：m_notEmpty 是给消费者（工作线程）用的，m_notFull 是给生产者（提交任务的线程）用的，各等各的，减少无效唤醒；
• 超时等待：用 wait_for 替代无限阻塞，避免线程“死等”；
• 移动语义：Put(T &&task) 支持右值引用，大任务拷贝成本一下就降下来了；
• 异常安全：靠 unique_lock 管理锁，就算抛异常也能保证锁自动释放。

2.2 缓存线程池核心（CachedThreadPool）

线程池的核心逻辑都集中在线程管理和任务调度这两块，下面把主要模块拆开看。

2.2.1 核心成员变量

class CachedThreadPool {
    public:
        using Task = std::function;
    private:
        static int KeepAliveTime;
        std::unordered_map> m_threadgroup;
        int m_coreThreadSize = 4;
        int m_maxThreadSize = std::thread::hardware_concurrency();
        mutable std::mutex m_mutex;
        std::condition_variable m_threadExit;
        std::atomic m_idleThreadSize = 0;
        std::atomic m_curThreadSize = 0;
        tulun::SyncQueue m_queue;
        std::atomic m_running = false;
        std::once_flag m_flag;
};

2.2.2 CachedThreadPool.hpp

#include "SyncQueue_1.hpp"
#include 
#include 

2.2.3 线程池初始化（Start）

初始化的时候，会先创建一批核心线程，它们相当于常驻的“主力部队”：

void CachedThreadPool::Start(int numthreads)
    {
        m_running = true;
        m_curThreadSize = numthreads;
        for (int i = 0; i < numthreads; ++i)
        {
            auto tha = std::make_shared(&CachedThreadPool::RunInThread, this);
            std::thread::id tid = tha->get_id();
            m_threadgroup.emplace(tid, tha);
            ++m_idleThreadSize;
        }
    }

2.2.4 工作线程逻辑（RunInThread）

工作线程的日常就是“循环拿任务、干活，要是闲太久就主动退场”：

int CachedThreadPool::KeepAliveTime = 60;
void CachedThreadPool::RunInThread()
    {
        auto tid = std::this_thread::get_id();
        auto startTime = std::chrono::high_resolution_clock().now();
        while (m_running)
        {
            Task task;
            if (m_queue.Empty())
            {
                auto now = std::chrono::high_resolution_clock().now();
                auto intervalTime =
                    std::chrono::duration_cast(now - startTime).count();
                std::lock_guard locker(m_mutex);
                if (intervalTime >= KeepAliveTime && m_curThreadSize > m_coreThreadSize)
                {
                    m_threadgroup.find(tid)->second->detach();
                    m_threadgroup.erase(tid);
                    m_curThreadSize--;
                    m_idleThreadSize--;
                    return;
                }
            }
            if (m_queue.Take(task) != 0)
            {
                continue;
            }
            if (m_running && task)
            {
                m_idleThreadSize--;
                task();
                m_idleThreadSize++;
                startTime = std::chrono::high_resolution_clock().now();
            }
        }
    }

2.2.5 动态创建线程（newThread）

一旦发现空闲线程不够用了，同时当前线程数还没达到上限，就会立刻创建新线程：

void CachedThreadPool::newThread()
    {
        std::lock_guard locker(m_mutex);
        if (m_idleThreadSize <= 0 && m_curThreadSize < m_maxThreadSize)
        {
            auto tha = std::make_shared(&CachedThreadPool::RunInThread, this);
            std::thread::id tid = tha->get_id();
            m_threadgroup.emplace(tid, tha);
            m_idleThreadSize++;
            m_curThreadSize++;
        }
    }

2.2.6 任务提交接口

任务提交分两种：一种是普通任务（没有返回值），另一种是异步任务（需要拿回结果）。

普通任务提交：

void CachedThreadPool::AddTask(Task &&task)
    {
        if (m_queue.Put(std::forward(task)) != 0)
        {
            LOG_INFO << "task()";
            task();
        }
        newThread();
    }
    void CachedThreadPool::AddTask(const Task &task)
    {
        if (m_queue.Put(task) != 0)
        {
            LOG_INFO << "task()";
            task();
        }
        newThread();
    }

异步任务提交（带返回值）：通过 packaged_task 和 future 这对搭档，把任务封装起来，调用方可以异步拿到执行结果：

template 
        auto submit(Func &&func, Args &&...args)
        {
            using RetType = decltype(std::invoke(std::forward(func), std::forward(args)...));
            auto task = std::make_shared>(
                [func = std::forward(func), ... args = std::forward(args)]() mutable
                {
                    return std::invoke(func, args...);
                });
            std::future result = task->get_future();
            if (m_queue.Put([task]() -> void
                            { (*task)(); }) != 0)
            {
                LOG_ERROR << "Add task run task";
                (*task)();
            }
            newThread();
            return result;
        }

2.2.7 优雅停止（StopThreadGroup）

停止线程池的时候，必须等所有任务都处理完，然后再统一回收所有线程：

void CachedThreadPool::StopThreadGroup()
    {
        m_queue.WaitQueueEmptyStop();
        m_coreThreadSize = 0;
        KeepAliveTime = 0;
        std::unique_lock locker(m_mutex);
        while (!m_threadgroup.empty())
        {
            m_threadExit.wait_for(locker, std::chrono::milliseconds(100));
        }
        m_running = false;
    }
void CachedThreadPool::Stop()
    {
        std::call_once(m_flag, std::bind(&CachedThreadPool::StopThreadGroup, this));
    }

2.2.8 构造+析构

CachedThreadPool::CachedThreadPool(int initnumthreads, int taskqueuesize)
        : m_coreThreadSize(initnumthreads),
          m_queue(taskqueuesize)
    {
        Start(initnumthreads);
    }
CachedThreadPool::~CachedThreadPool()
    {
        if (m_running)
        {
            Stop();
        }
    }

三、缓存线程池 vs 固定线程池：核心特性对比

从对比可以看出：

• FixedThreadPool 更擅长稳定的 CPU 密集型任务，省去了线程频繁创建销毁的开销；
• CachedThreadPool 则适合短期、并发波动大的 IO 密集型任务，靠动态扩缩容来平衡性能和资源。

四、使用场景与最佳实践

4.1 缓存线程池的适用场景

结合它的特性，下面这几类场景特别适合用缓存线程池：

1. 大量短期任务：任务来了就尽量创建线程去执行，空闲线程还能复用，避免反复创建销毁的额外开销；
2. 任务响应快速：任务量上来时能快速创建新线程，减少任务的排队等待时间；
3. 无需限制线程数量：只要内存扛得住，可以动态创建新线程；
4. 短期任务高并发：任务完成后，线程池还会保留一批空闲线程，为下一波任务做好准备。

4.2 最佳实践与避坑指南

缓存线程池并不是“万能药”，在高负载场景下也有自己的风险，工程使用时下面几点要注意：

1. 警惕无界队列风险：无界队列很容易导致内存溢出或任务延迟飙升，建议换成有界队列；
2. 控制最大线程数：默认的上限太大，可能导致系统资源被耗尽，最好自己设个合理的最大值；
3. 避免长时间阻塞任务：缓存线程池不适合执行耗时长的任务，否则后续任务会堆积，线程数也会失控地膨胀；
4. 推荐使用 ThreadPoolExecutor：不管是 Ja va 还是 C++，都可以直接用 ThreadPoolExecutor（或者它的 C++ 实现），通过自定义核心线程数、最大线程数、队列大小和拒绝策略来做更精细的控制；
5. 扩展钩子函数：可以重写 beforeExecute/afterExecute 这类钩子，在任务执行前后加点自定义逻辑，比如打日志、做监控；
6. 线程工厂定制：通过自定义 ThreadFactory 给线程命名、设置守护属性，排查问题的时候能省不少事；
7. 动态调整池大小：可以实现一个动态线程池，运行时根据任务量和 CPU 使用率来调整核心线程数与最大线程数。

五、缓存线程池的使用示例

下面是一个实际用的例子，看看缓存线程池是怎么处理大规模任务的（这里是随机数生成）：

#include 
#include 
using namespace std;
#include "Timestamp.hpp"
#include "Logger.hpp"
#include "CachedThreadPool.hpp"
static const size_t row = 10000;
static const size_t col = 100000;
std::vector> iveca, ivecb, ivecc, ivecd;
void RandInit(std::vector &ivec)
{
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution dis(0, 100000);
    for (int i = 0; i < col; ++i)
    {
        ivec.push_back(dis(gen));
    }
}
void SortVec(std::vector &ivec)
{
    std::sort(ivec.begin(), ivec.end());
}
void Sa veFile(std::vector &ivec, const std::string &filename)
{
    FILE *pf = fopen(filename.c_str(), "a");
    if (nullptr == pf)
    {
        LOG_FATAL << "fopen fail \n";
        exit(EXIT_FAILURE);
    }
    for (int i = 0; i < col; ++i)
    {
        fprintf(pf, "%d ", ivec[i]);
    }
    fprintf(pf, "\n---------------------\n");
    fclose(pf);
    pf = nullptr;
}
void test_funb()
{
    srand(time(nullptr));
    cout << "4 线程测试...." << endl;
    tulun::Timestamp start, end;
    start = tulun::Timestamp::Now();
    {
        ivecb.resize(row);
        tulun::CachedThreadPool mypool(4, 500);
        for (int i = 0; i < row; ++i)
        {
            ivecb[i].reserve(col);
            mypool.AddTask(std::bind(RandInit, std::ref(ivecb[i])));
            if ((i + 1) % 5000 == 0)
            {
                mypool.PrintInfo();
            }
        }
    }
    end = tulun::Timestamp::Now();
    cout << "RandInit: " << (tulun::diffMicro(end, start)) / 1000000 << "." << (tulun::diffMicro(end, start)) % 1000000 << endl;
    start = tulun::Timestamp::Now();
    {
        tulun::CachedThreadPool mypool(4, 500);
        for (int i = 0; i < row; ++i)
        {
            mypool.AddTask(std::bind(SortVec, std::ref(ivecb[i])));
            if ((i + 1) % 5000 == 0)
            {
                mypool.PrintInfo();
            }
        }
    }
    end = tulun::Timestamp::Now();
    cout << "SortVec : " << (tulun::diffMicro(end, start)) / 1000000 << "." << (tulun::diffMicro(end, start)) % 1000000 << endl;
    start = tulun::Timestamp::Now();
    {
        tulun::CachedThreadPool mypool(4, 500);
        for (int i = 0; i < row; ++i)
        {
            mypool.AddTask(std::bind(Sa veFile, std::ref(ivecb[i]), std::string("ivecb.txt")));
            if ((i + 1) % 5000 == 0)
            {
                mypool.PrintInfo();
            }
        }
    }
    end = tulun::Timestamp::Now();
    cout << "Sa veFile: " << (tulun::diffMicro(end, start)) / 1000000 << "." << (tulun::diffMicro(end, start)) % 1000000 << endl;
    return;
}
void funa()
{
    for (int i = 0; i < 1000; ++i)
        ;
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
void funb()
{
    for (int i = 0; i < 100; ++i)
        ;
    std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100));
}
void test_func()
{
    {
        tulun::CachedThreadPool mypool(4, 500);
        for (int i = 0; i < 1000; ++i)
        {
            mypool.AddTask(funa);
            if ((i + 1) % 100 == 0)
            {
                mypool.PrintInfo();
            }
        }
        cout << "###############################" << endl;
        for (int i = 0; i < 1000; ++i)
        {
            mypool.AddTask(funb);
            if ((i + 1) % 10 == 0)
            {
                mypool.PrintInfo();
            }
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        mypool.PrintInfo();
    }
}
int main()
{
    tulun::Logger::setLogLevel(tulun::LOG_LEVEL::INFO);
    test_func();
    return 0;
}

六、核心设计亮点与优化方向

6.1 设计亮点

• 原子变量保证线程安全：m_idleThreadSize、m_curThreadSize 这些字段全用的 std::atomic，能有效避免锁竞争；
• 超时机制避免永久阻塞：任务队列的 wait_for 加上线程回收的超时判断，系统鲁棒性一下就上来了；
• 移动语义减少拷贝：任务提交支持右值引用，大任务拷贝开销显著降低；
• 降级策略：任务队列满了之后，直接在提交线程里执行任务，至少不会丢任务；
• 异步任务支持：借助 packaged_task 和 future，带返回值的任务也能轻松提交。

6.2 可优化方向

• 线程回收机制：当前用了 detach 来回收线程，可以考虑改成 join，结合线程池销毁逻辑，更安全；
• 任务优先级：扩展任务队列，让它支持优先级，高优先级的任务可以插队执行；
• 线程局部存储：给工作线程加上 TLS（线程局部存储），减少线程间的资源竞争；
• 监控与统计：增加任务执行耗时、线程创建/回收次数等统计指标，方便做性能调优；
• 异常捕获：工作线程执行任务时应该捕获异常，避免因为一个任务崩溃就导致整个线程退出。

七、总结

缓存线程池在高并发场景下，可以说是一个很高效的线程管理方案。它的核心价值就在于用动态扩缩容来平衡性能与资源消耗：

• 用线程安全的任务队列把任务提交和执行拆开，互不干扰；
• 核心线程常驻，临时线程按需创建、超时回收；
• 原子变量和条件变量配合起来，保证多线程下的状态同步；
• 降级策略和超时机制一起上，系统的鲁棒性更有保障。

不过还是得强调一句：缓存线程池不是“银弹”。在高负载、长时间任务的场景下，用起来要格外小心。更推荐的做法是参考 ThreadPoolExecutor 的设计思想，通过自定义参数来实现一个更可控的线程池。