本实现基于 C20 标准摒弃了传统的常规标志位与易错组件。以下是核心代码段的设计动机与原理为什么使用 std::jthread 而不是 std::threadstd::jthread 提供了自动汇合和协作式中断机制。它的析构函数会自动发出停止请求并阻塞等待线程结束彻底消除了传统 std::thread 忘记调用 join() 导致程序崩溃的隐患并且代码更简洁安全性更高。这看着像GC机制自动销毁往往伴随着一些代价虽然代价很小但在性能要求极高的场景极低延迟与高频交易系统硬实时系统高频短生命周期并发任务上面会体现出来。为什么不需要自定义 bool stop 标志位工作线程直接接收 std::stop_token。析构函数中遍历调用 request_stop()或者直接依赖 jthread 的自动机制st.stop_requested() 会安全地接管停止信号。这使得中断逻辑与操作系统底层的线程机制绑定更深代码更简洁。为什么条件变量等待的判断逻辑是 st.stop_requested() || !tasks.empty()且随后判断 st.stop_requested() tasks.empty()这是为了实现优雅降级。收到停止信号后线程不会立刻暴力退出而是会继续消耗队列中残留的任务。只有当既收到停止请求且任务队列被彻底清空时线程才会 return 结束。为什么必须用 std::shared_ptr 包装 std::packaged_task这是 C 线程池最经典的坑。任务队列的类型是 std::functionvoid()C 标准要求 std::function 内部持有的对象必须是可拷贝的CopyConstructible。但 std::packaged_task 作为一个管理异步状态的类是仅可移动的Move-Only。通过 std::make_shared 将其放到堆上Lambda 表达式按值捕获这个 shared_ptr发生拷贝从而完美满足了 std::function 的编译要求。为什么使用 requires std::invocableFunc, Args...利用 C20 的 Concepts 机制进行编译期约束。如果调用者传入了参数类型不匹配的函数编译器会在 enqueue 调用处直接报错而不是在深层模板实例化时抛出几百行的乱码错误极大提升了 API 的易用性。为什么 Lambda 捕获要写成 [fstd::forwardFunc(f), ...argsstd::forwardArgs(args)...]() mutable这是 C20 引入的带初始化器的包扩展捕获Pack expansion in lambda init-capture。它结合了完美转发确保任务入队时参数的左值/右值属性被完整保留到实际执行的那一刻。mutable 关键字则是因为 std::invoke 在执行某些带有内部状态的仿函数时可能需要修改捕获进来的按值变量。为什么入队时要判断 pool[0].get_stop_token().stop_requested()防止“死后入队”。当线程池生命周期即将结束析构函数已触发时如果其他线程还在尝试往池子里塞任务这些任务将永远不会被执行。直接抛出 std::runtime_error 可以在运行时尽早暴露时序 Bug。#include iostream #include thread #include condition_variable #include mutex #include functional #include vector #include queue #includefuture #includememory #include concepts #includestdexcept class ThreadPool { public: explicit ThreadPool(size_t n std::thread::hardware_concurrency()) { pool.reserve(n); for(size_t i0;in;i){ pool.emplace_back([this](std::stop_token st){ while(!st.stop_requested()){ std::functionvoid() task; { std::unique_lockstd::mutex lock(mu); c_v.wait(lock,[this,st](){return st.stop_requested()||!tasks.empty();}); if(st.stop_requested()tasks.empty())return; taskstd::move(tasks.front()); tasks.pop(); } task(); } }); } } ~ThreadPool() { for (auto t : pool) { t.request_stop(); } c_v.notify_all(); } templatetypename Func,typename ...Args requires std::invocableFunc,Args... auto enqueue(Func f,Args ...args)-std::futurestd::invoke_result_tFunc,Args...{ using retTypestd::invoke_result_tFunc,Args...; auto taskstd::make_sharedstd::packaged_task(retType()) ([fstd::forwardFunc(f),...argsstd::forwardArgs(args)...]()mutable{ return std::invoke(f,std::move(args)...);}); std::futureretType restask-get_future(); { std::unique_lockstd::mutexlock(mu); if (!pool.empty() pool[0].get_stop_token().stop_requested()) { throw std::runtime_error(enqueue on stopped ThreadPool); } tasks.emplace([task](){(*task)();}); } c_v.notify_one(); return res; } private: std::mutex mu; std::condition_variable c_v; std::vectorstd::jthread pool; std::queuestd::functionvoid() tasks; };
