现代 C++ 异步编程：从零实现一个高性能 ThreadPool (C++20 深度实践)

在高性能 C 开发中线程池是绕不开的核心基础设施。随着 C20 标准的普及我们能够以更简洁、更安全的方式实现一个生产级的线程池。本文将带你深度剖析一个基于std::jthread的线程池实现并探讨其背后的架构思考与内存管理机制。1. 核心代码实现 (C20 版)这个线程池利用了 C20 的 RAII 线程管理特性彻底告别了手动join的时代。#pragmaonce#includecondition_variable#includefunctional#includefuture#includememory#includemutex#includequeue#includestdexcept#includethread#includetype_traits#includeutility#includevectorclassThreadPool{public:explicitThreadPool(std::size_t threads){if(threads0){throwstd::invalid_argument(ThreadPool requires at least one worker);}workers.reserve(threads);for(std::size_t i0;ithreads;i){// std::jthread 会自动管理线程生命周期。//// jthread 析构时会// 1. request_stop()// 2. join()//// stop_token 是 worker 获取停止请求的方式。workers.emplace_back([this](std::stop_token st){// 不要写//// while (!st.stop_requested())//// 因为我们的策略是//// 收到 stop// ↓// 不再等待新任务// ↓// 把已经在队列中的任务执行完// ↓// worker 退出//// 所以这里使用无限循环由下面的逻辑决定什么时候退出。for(;;){std::functionvoid()task;{std::unique_lockstd::mutexlock(queue_mutex);// 等待两个条件之一//// 1. tasks 非空// 2. stop_token 收到 stop 请求//// condition_variable_any 支持 stop_token// 因此 request_stop() 可以唤醒这里的 worker。//// wait() 返回//// true - predicate 满足即 tasks 非空// false - stop_token 收到了 stop 请求boolokcondition.wait(lock,st,[this]{return!tasks.empty();});// stop 请求并且任务队列已经为空//// stop true// tasks.empty() true//// 说明没有任何任务需要继续处理可以退出。if(!oktasks.empty()){return;}// 到这里有两种可能//// 1. 正常收到任务// 2. 收到 stop但队列里还有任务//// 两种情况都需要取出任务执行。taskstd::move(tasks.front());tasks.pop();}// 非常重要//// 不要持有 queue_mutex 执行任务。//// 如果任务执行时间很长而 mutex 一直被占用// 其他 worker 就无法从任务队列取任务。//// 所以//// 加锁// ↓// 取任务// ↓// 解锁// ↓// 执行任务task();}});}}~ThreadPool()default;templateclassF,class...Argsautoenqueue(Ff,Args...args)-std::futurestd::invoke_result_tF,Args...{// 获取任务函数的返回类型。//// 例如//// int foo(int);//// enqueue(foo, 10)//// return_type intusingreturn_typestd::invoke_result_tF,Args...;// packaged_task 用于//// worker 执行任务// ↓// 保存返回值// ↓// future 获取返回值//// packaged_task 本身是 move-only。//// 所以这里使用 shared_ptr 管理它的生命周期。autotaskstd::make_sharedstd::packaged_taskreturn_type()(std::bind(std::forwardF(f),std::forwardArgs(args)...));// future 是调用者获取任务执行结果的方式。std::futurereturn_typeresulttask-get_future();{std::unique_lockstd::mutexlock(queue_mutex);// tasks 的统一类型是//// std::functionvoid()//// 不管用户提交的是//// int foo()// void bar()// std::string baz()//// 最终都包装成//// void()//// worker 不需要知道任务的真实类型。tasks.emplace([task]{(*task)();});}// mutex 已经释放之后再通知 worker。//// notify_one() 唤醒一个正在等待任务的 worker。condition.notify_one();returnresult;}private:/* * 成员变量的声明顺序非常重要。 * * C 对成员变量的析构顺序是 * * 声明顺序的逆序 * * 所以这里实际上会按照 * * workers * ↓ * tasks * ↓ * condition * ↓ * queue_mutex * * 的顺序析构。 * * workers 必须最后声明。 * * 原因 * * worker 线程里面会访问 * * queue_mutex * condition * tasks * * 因此必须保证 worker 全部结束之后 * 这些对象才开始析构。 */// 保护 tasks。std::mutex queue_mutex;// condition_variable_any 支持 stop_token。std::condition_variable_any condition;// 等待执行的任务。std::queuestd::functionvoid()tasks;// 必须最后声明。//// 析构时 workers 最先被析构//// ~jthread()// ↓// request_stop()// ↓// worker 被唤醒// ↓// worker 退出// ↓// join()//// 等所有 worker 结束以后// tasks / condition / queue_mutex 才会析构。std::vectorstd::jthreadworkers;};2. 深度剖析为什么不直接存packaged_task这是初学者常有的疑问既然std::packaged_task已经封装了任务为什么队列里存的是std::functionvoid()类型擦除 (Type Erasure)std::packaged_taskR(Args...)是一个强类型模板它的类型包含了返回值R。如果任务 A 返回int任务 B 返回string它们的类型是不同的。std::queue只能存储相同类型的对象。解决方案我们利用std::functionvoid()作为“通用信封”。在enqueue内部我们写了一个 Lambda 表达式[task](){ (*task)(); }。无论原始任务返回什么这个 Lambda 永远是void()类型。这在设计模式中被称为类型擦除它抹平了不同任务之间的差异。3. 内存管理智能指针的“接力”在异步环境下最怕的是“任务还没跑对象先没了”。本实现通过std::shared_ptr和 Lambda 捕获完美解决了生存周期问题。诞生在enqueue里通过make_shared创建任务。此时计数为 1。接力Lambda 表达式通过按值捕获[task]。这意味着shared_ptr被拷贝了一份存进了 Lambda 闭包中。此时计数为 2。入队Lambda 被存入std::function并进入队列。即使enqueue函数返回闭包依然拉着shared_ptr。销毁工作线程取出并执行完任务后std::function被销毁引用计数降为 0任务对象内存自动释放。这种“引用计数闭包捕获”的机制保证了任务只要在队列中内存就绝对安全。4. C20 的技术红利std::jthread 与 RAII传统的std::thread在析构时如果不join或detach会导致进程异常退出。std::jthread引入了 RAII资源获取即初始化机制它在析构时会自动发出停止请求并等待线程结束使得线程池的析构函数极其简洁。协作式中断 (stop_token)通过std::stop_token我们不再需要手动维护一个bool stop标志。condition.wait(lock, st, pred)更是神来之笔当jthread准备停止时它会自动唤醒所有阻塞在条件变量上的线程。这比以前手动notify_all()要健壮得多。5. 如何使用intmain(){ThreadPoolpool(4);// 提交带返回值的异步任务autofuturepool.enqueue([](intx){returnx*x;},10);// 获取结果 (会阻塞直到任务完成)std::coutResult: future.get()std::endl;return0;}6. 总结实现一个线程池不难但实现一个既类型安全又内存无忧的线程池需要对 C 的底层机制有深刻理解。std::functionvoid()解决了异质任务的存储问题。std::shared_ptr解决了异步任务的生命周期问题。std::jthread解决了线程资源的回收问题。这个短小精悍的实现正是 C 现代化的魅力所在。版权声明本文采用 CC BY-SA 4.0 协议转载请注明出处。