1. 项目概述最近在整理一些C项目时发现很多同学对线程池的实现原理和使用场景存在不少疑问。线程池这个听起来有点“高级”的概念其实在C并发编程中扮演着至关重要的角色。简单来说它就是一个预先创建好一批线程的“池子”当有任务需要执行时就从池子里分配一个空闲线程去干活干完了再放回池子里待命而不是每次来了任务都临时创建、用完再销毁线程。这种模式在高并发、高性能的服务端程序、游戏服务器、数据处理框架中几乎是标配。为什么我们需要自己动手实现一个市面上不是有现成的库吗比如Boost.Asio的thread_pool或者Qt的QThreadPool。原因很简单第一理解原理。自己从零实现一遍对条件变量、互斥锁、任务队列这些并发编程的核心组件会有刻骨铭心的理解。第二轻量可控。很多项目可能不希望引入庞大的Boost或Qt库一个百来行代码、功能清晰的自研线程池既满足了需求又避免了依赖膨胀。第三定制化。你可以根据自己的业务场景轻松添加优先级队列、动态线程调整、任务取消等高级特性。这篇文章我就以一个从业多年的C工程师的视角带你从零开始手把手实现一个功能完备、工业可用的ThreadPool。我们会从最基础的版本开始逐步迭代加入异常安全、优雅关闭、返回值获取等特性并深入探讨每个设计决策背后的“为什么”。无论你是正在准备面试的校招生还是想夯实基础的职场人相信这篇长文都能给你带来实实在在的收获。2. 线程池的核心设计与架构拆解在动手写代码之前我们必须先把线程池的“骨架”和“运转逻辑”想清楚。一个典型的线程池主要由以下几个核心部件构成2.1 核心组件与职责划分任务队列 (Task Queue)这是一个线程安全的队列用于存放所有待执行的任务。生产者线程主线程或其他工作线程向队列中提交任务消费者线程线程池中的工作线程从队列中取出任务执行。队列的线程安全是重中之重必须用锁来保护。工作线程组 (Worker Threads)一组预先创建好的、处于运行状态的线程。它们的工作就是不断地、循环地从任务队列中取出任务并执行。如果没有任务它们应该进入等待状态而不是忙等以避免浪费CPU资源。同步机制 (Synchronization Primitives)主要是互斥锁 (std::mutex)和条件变量 (std::condition_variable)。互斥锁用于保护对共享资源任务队列的并发访问确保同一时间只有一个线程能进行入队或出队操作。条件变量这是实现高效等待/通知机制的关键。当任务队列为空时工作线程通过条件变量进入等待状态当有新任务入队时提交任务的线程通过条件变量通知唤醒一个或多个等待中的工作线程。停止标志 (Stop Flag)一个原子布尔变量 (std::atomicbool)用于通知所有工作线程“该收工了”。当线程池需要关闭时设置这个标志并通知所有等待中的线程让它们优雅退出循环。2.2 线程池的工作流程整个线程池的生命周期可以概括为以下几个阶段初始化 (Initialization)根据用户指定的数量或根据CPU核心数自动计算创建N个工作线程。每个线程启动后立即进入一个“等待-执行”循环。任务提交 (Submission)用户通过submit函数提交一个可调用对象函数、Lambda、函数对象等及其参数。线程池内部将这个调用包装成一个“任务单元”放入任务队列然后通过条件变量通知一个等待中的工作线程。任务执行 (Execution)被唤醒的工作线程从任务队列头部取出一个任务释放队列锁然后执行这个任务。执行完毕后线程再次尝试从队列中取任务如果队列为空则再次进入等待状态。关闭与清理 (Shutdown Cleanup)当需要销毁线程池时首先设置停止标志然后通过条件变量notify_all()唤醒所有正在等待的工作线程。每个工作线程被唤醒后检查到停止标志为真便会退出循环。主线程随后调用每个工作线程的join()方法等待它们全部执行完毕完成资源清理。2.3 设计决策为什么选择这样的架构为什么用队列而不是其他数据结构队列FIFO先进先出是最自然、最公平的任务调度策略。先来的任务先被执行符合大多数场景的预期。当然你也可以实现基于优先级的队列如使用std::priority_queue但这会增加复杂度我们会在进阶部分讨论。为什么工作线程要循环等待而不是执行一次就退出创建和销毁线程的开销非常大涉及系统调用和内核资源分配。线程池的核心思想就是“复用”。让线程长期存活循环执行任务避免了频繁创建销毁的巨大开销这是提升性能的关键。为什么用条件变量而不是让线程忙等 (busy-waiting)忙等是指线程不断循环检查队列是否为空 (while (tasks_.empty()) {})。这会白白消耗一个完整的CPU核心导致CPU使用率飙升性能急剧下降。条件变量则让线程在队列为空时主动让出CPU进入休眠状态直到被其他线程唤醒这才是高效的做法。为什么停止标志要用std::atomic因为这个标志会被多个线程同时读写主线程写所有工作线程读。std::atomic确保了对此变量的读写操作是原子的并且会生成正确的内存屏障指令防止编译器或CPU进行有害的重排序从而保证所有工作线程能及时、正确地看到停止信号。3. 基础版本线程池实现详解理论说再多不如一行代码。我们现在就来实现第一个可运行的、最基础的线程池版本。这个版本实现了核心功能但为了清晰暂时省略了异常处理和任务返回值获取。3.1 头文件与成员变量#include vector #include queue #include thread #include mutex #include condition_variable #include functional #include atomic class ThreadPool { public: explicit ThreadPool(size_t num_threads); ~ThreadPool(); // 提交一个无返回值的任务 templateclass F void enqueue(F task); // 停止线程池等待所有已入队任务完成 void stop(); private: // 工作线程的主循环函数 void worker_thread(); std::vectorstd::thread workers_; // 工作线程容器 std::queuestd::functionvoid() tasks_; // 任务队列 std::mutex queue_mutex_; // 保护任务队列的互斥锁 std::condition_variable condition_; // 用于线程等待/通知的条件变量 std::atomicbool stop_{false}; // 停止标志 bool joining_{false}; // 标记是否正在等待线程结束 };成员变量解析workers_: 存储std::thread对象。在构造函数中创建在析构函数中join。tasks_: 任务队列。存储类型为std::functionvoid()的可调用对象代表一个无参数、无返回值的任务。queue_mutex_: 任何对tasks_的访问push,pop,empty都必须先锁住这个互斥锁。condition_: 工作线程在任务队列为空时在此条件变量上等待enqueue提交新任务时通过此变量通知一个等待线程。stop_: 原子布尔量。为true时所有工作线程应退出循环。joining_: 一个辅助标志用于防止在stop()函数中重复join线程这不是原子的因为它只在stop()函数内由主线程访问。3.2 构造函数与工作线程函数ThreadPool::ThreadPool(size_t num_threads) { if (num_threads 0) { num_threads std::thread::hardware_concurrency(); if (num_threads 0) num_threads 1; // 硬件并发数未知至少1个 } workers_.reserve(num_threads); for (size_t i 0; i num_threads; i) { // 创建线程并立即执行worker_thread成员函数 workers_.emplace_back(ThreadPool::worker_thread, this); } } void ThreadPool::worker_thread() { while (true) { std::functionvoid() task; { // 1. 获取队列锁 std::unique_lockstd::mutex lock(queue_mutex_); // 2. 等待条件停止 或 队列非空 // condition_.wait会原子地释放锁并进入等待被唤醒后重新获取锁 condition_.wait(lock, [this]() { return stop_.load() || !tasks_.empty(); }); // 3. 检查退出条件 if (stop_.load() tasks_.empty()) { return; // 停止标志为真且队列已空线程退出 } // 4. 从队列中取出任务 task std::move(tasks_.front()); tasks_.pop(); } // 锁的作用域结束自动释放锁 // 5. 执行任务在锁外执行避免长时间持有锁阻塞其他线程 task(); } }worker_thread函数是核心中的核心我们来逐行分析std::unique_lock在构造时锁定互斥锁在析构时自动释放。它比std::lock_guard更灵活可以手动解锁是配合条件变量的标准选择。condition_.wait(lock, predicate)这是条件变量的标准用法。它会检查predicateLambda表达式是否为真。如果为真即stop_为真或队列非空则直接返回继续执行。如果为假则原子地释放lock并将当前线程挂起进入等待状态直到被其他线程的condition_.notify_one()或condition_.notify_all()唤醒。被唤醒后它会重新获取锁然后再次检查predicate。只有predicate为真时wait才会返回。这种“唤醒后检查”的模式称为“虚假唤醒”防护是必须的。被唤醒后我们再次确认退出条件如果线程池已停止并且任务队列已空那么此工作线程的使命就完成了直接返回线程函数结束线程将自然结束。如果不需要退出就从队列头部取出一个任务。使用std::move避免不必要的拷贝。关键点在锁的作用域外执行任务。任务执行时间可能很长如果持有锁执行其他所有线程都无法访问任务队列提交或获取任务并发性能将荡然无存。所以取出任务后立即释放锁让其他线程可以继续操作队列。3.3 任务提交与停止函数templateclass F void ThreadPool::enqueue(F task) { { std::lock_guardstd::mutex lock(queue_mutex_); if (stop_.load()) { throw std::runtime_error(enqueue on stopped ThreadPool); } tasks_.emplace(std::forwardF(task)); } // lock_guard析构自动释放锁 condition_.notify_one(); // 通知一个等待中的工作线程 } void ThreadPool::stop() { if (joining_.exchange(true)) { return; // 防止重复调用stop } stop_.store(true); condition_.notify_all(); // 唤醒所有等待的线程让它们检查stop_标志 for (std::thread worker : workers_) { if (worker.joinable()) { worker.join(); } } workers_.clear(); } ThreadPool::~ThreadPool() { // 析构函数自动调用stop确保资源清理 if (!joining_) { stop(); } }enqueue函数解析使用std::lock_guard保护入队操作。如果线程池已停止则拒绝新任务抛出异常。这是一个重要的安全设计。使用std::forwardF(task)进行完美转发保持传入任务的值类别左值/右值。入队后调用condition_.notify_one()。注意这里是在锁释放之后才通知。如果在锁内通知被唤醒的线程会立刻尝试获取锁但锁还被当前线程持有导致它再次阻塞造成不必要的上下文切换。这是一个常见的性能优化点。stop函数解析使用std::atomic::exchange设置joining_标志并检查其旧值确保stop()函数只生效一次。设置stop_标志为true。notify_all()唤醒所有在condition_上等待的工作线程。这是必须的因为我们要让所有线程都看到停止信号并退出。遍历所有工作线程调用join()等待它们执行完毕。joinable()检查是必要的防止对已经join过的线程再次操作。析构函数调用stop()实现了RAII资源获取即初始化用户即使忘记手动调用stop也能保证线程安全退出。3.4 基础版本的使用示例与局限#include iostream #include chrono int main() { ThreadPool pool(4); // 创建4个线程的池子 // 提交10个任务 for (int i 0; i 10; i) { pool.enqueue([i] { std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟耗时操作 std::cout Task i executed by thread std::this_thread::get_id() std::endl; }); } std::this_thread::sleep_for(std::chrono::seconds(2)); // 等待任务执行 pool.stop(); // 手动停止析构函数也会调用 std::cout All tasks completed.\n; return 0; }基础版本的局限性无法获取任务返回值enqueue函数返回void任务执行的结果无法传递回调用者。异常会崩溃如果任务在执行过程中抛出异常这个异常会传播到工作线程的task()调用处导致整个工作线程异常终止程序可能崩溃。任务类型单一只接受void()签名的任务。关闭策略固定stop()会等待所有已入队任务完成但无法实现“立即停止并丢弃未执行任务”的策略。接下来我们就针对这些局限性一步步打造一个工业级的线程池。4. 进阶版本支持返回值与异常安全在实际项目中我们经常需要获取异步任务的执行结果并且要确保任务中的异常不会导致整个线程池崩溃。C标准库中的std::future和std::packaged_task正是为此而生的完美搭档。4.1 使用std::packaged_task包装任务std::packaged_task可以将任何可调用对象包装起来并将其执行结果或异常存储在一个与之关联的std::future对象中。我们可以修改任务队列存储std::packaged_taskvoid()吗不行因为packaged_task是不可拷贝的。解决方案是使用类型擦除并用std::function包装一个std::packaged_task的移动操作或者更直接地使用std::packaged_task的基类std::function无法处理我们需要自己定义一个可移动、可调用的任务类型。一个更简洁的方法是使用std::packaged_task并搭配std::unique_ptr来管理生命周期。但这里我介绍另一种常见模式定义一个内部的TaskWrapper基类然后通过派生类模板来保存任意类型的packaged_task。为了聚焦核心我们采用一个更直观的方法修改enqueue函数使其返回一个std::future并在内部使用std::packaged_task和std::shared_ptr来共享任务状态。4.2 改进的submit函数实现我们将enqueue改名为更贴切的submit并让它支持任意可调用对象和参数。class ThreadPool { public: // ... 其他成员 ... // 新的提交函数返回std::future以获取结果 templateclass F, class... Args auto submit(F f, Args... args) - std::futuredecltype(f(args...)); }; templateclass F, class... Args auto ThreadPool::submit(F f, Args... args) - std::futuredecltype(f(args...)) { // 推导任务返回类型 using return_type decltype(f(args...)); // 创建一个packaged_task绑定函数和参数。 // 使用std::bind将函数和参数打包成一个void()的可调用对象。 // 用shared_ptr管理确保任务对象在异步环境中存活。 auto task_ptr std::make_sharedstd::packaged_taskreturn_type()( std::bind(std::forwardF(f), std::forwardArgs(args)...) ); // 从packaged_task获取future对象用于后续获取结果 std::futurereturn_type res task_ptr-get_future(); { std::lock_guardstd::mutex lock(queue_mutex_); if (stop_.load()) { throw std::runtime_error(submit on stopped ThreadPool); } // 将任务包装成一个void()的lambda放入队列 // 这个lambda捕获shared_ptr执行packaged_task tasks_.emplace([task_ptr]() { (*task_ptr)(); }); } condition_.notify_one(); return res; // 返回future给调用者 }关键点解析返回值类型推导使用decltype和尾返回类型语法使submit函数能自动推导出任务f的返回类型。std::packaged_taskreturn_type()创建一个返回类型为return_type、无参数的packaged_task。packaged_task本身是可调用对象调用它就会执行绑定的函数。std::bind将用户传入的函数f和参数包args...绑定在一起生成一个无参数的可调用对象符合packaged_task的签名。使用std::forward进行完美转发。std::shared_ptr管理生命周期packaged_task不可拷贝但shared_ptr可以。通过shared_ptr捕获确保任务对象在Lambda表达式中一直有效直到被执行。Lambda包装任务队列tasks_存储的仍然是std::functionvoid()。我们用一个Lambda来包装packaged_task的调用(*task_ptr)()。当工作线程执行这个Lambda时就会触发packaged_task其结果或异常会自动存储到与之关联的future中。返回std::future调用者通过submit拿到一个future对象。他可以在任何需要的时候调用future.get()来获取结果如果任务未完成则会阻塞等待。如果任务执行中抛出了异常future.get()也会重新抛出该异常。4.3 异常安全与资源管理现在我们的线程池已经具备了异常安全的基础任务异常不会导致线程崩溃任务中的异常被packaged_task捕获并存储到future中。当工作线程执行(*task_ptr)()时即使内部函数抛出异常也只会导致这个packaged_task的执行终止并将异常状态存入future而不会将异常传播到工作线程函数之外。工作线程本身不会因为任务异常而退出循环。future.get()传播异常调用者通过future.get()获取结果时如果任务执行过程中发生了异常get()会重新抛出这个异常让调用者能够感知和处理。这符合C异常处理的常规模式。但是还有一个潜在问题如果用户在提交任务后线程池立即被stop()了而任务还在队列中未被执行那么与这个任务关联的future会怎样调用future.get()会一直阻塞因为任务永远不会被执行。我们需要在stop()时决定如何处理这些未执行的、带有future的任务。一种常见的策略是在stop()函数中除了设置标志和唤醒线程还可以选择清空任务队列。但清空队列会导致那些future永远得不到结果调用get()会阻塞。更合理的工业级实现需要更复杂的生命周期管理例如在ThreadPool析构时让所有未完成的future变为“就绪”状态并携带一个表示“任务被丢弃”的异常。这涉及到std::promise和std::future的更多高级用法超出了本文基础范围。在我们的实现中我们采用一种简单策略stop()会等待所有已入队任务执行完毕。这通过worker_thread中的退出条件if (stop_.load() tasks_.empty())来实现。这意味着调用stop()后线程池不会立即结束而是会等待队列清空。这样所有已提交任务的future最终都能得到结果或异常。4.4 进阶版本使用示例int compute_square(int x) { if (x 0) { throw std::invalid_argument(Negative input); } return x * x; } int main() { ThreadPool pool(2); std::vectorstd::futureint futures; // 提交一批计算任务 for (int i -2; i 5; i) { futures.push_back(pool.submit(compute_square, i)); } // 获取结果处理异常 for (size_t i 0; i futures.size(); i) { try { int result futures[i].get(); // 可能抛出异常 std::cout Task (i-2) result: result std::endl; } catch (const std::exception e) { std::cout Task (i-2) failed with: e.what() std::endl; } } pool.stop(); return 0; } // 可能的输出 // Task -2 failed with: Negative input // Task -1 failed with: Negative input // Task 0 result: 0 // Task 1 result: 1 // Task 2 result: 4 // Task 3 result: 9 // Task 4 result: 16这个例子展示了如何提交带参数和返回值的函数以及如何安全地处理任务中抛出的异常。5. 生产环境级优化与功能扩展基础功能和异常安全都有了但一个真正健壮的线程池还需要考虑更多。下面我们探讨几个常见的优化和扩展点。5.1 动态线程数量调整固定大小的线程池在某些场景下可能不是最优的。例如任务量波动很大有时需要更多线程来处理突发请求有时线程闲置又浪费资源。我们可以实现动态调整线程数量的功能。思路添加一个max_threads_和min_threads_成员变量。添加一个idle_threads_计数器原子变量记录当前空闲的线程数。修改worker_thread当线程从任务队列取不到任务时不是立即等待而是先检查空闲线程是否过多。如果超过min_threads_可以让该线程结束运行线程数减少。这需要更精细的条件变量控制。在submit时如果队列积压任务过多且当前线程数小于max_threads_可以动态创建新的线程。实现复杂度较高需要平衡性能和控制逻辑。一个简化版的思路是提供resize(size_t new_size)接口允许用户手动调整但内部实现需要安全地增加或减少工作线程。5.2 任务优先级调度默认的FIFO队列无法处理任务优先级。我们可以将std::queue替换为std::priority_queue并定义自己的任务优先级比较函数。struct PriorityTask { int priority; std::functionvoid() task; // 重载运算符使priority_queue成为最大堆优先级数字大的先出队 bool operator(const PriorityTask other) const { return priority other.priority; // 注意priority_queue默认是最大堆但比较用 } }; // 在ThreadPool中 std::priority_queuePriorityTask tasks_;提交任务时需要附带优先级参数。工作线程从优先队列中取任务。需要注意的是std::priority_queue的top()返回常量引用pop()不返回值我们需要先top()获取任务再pop()。5.3 优雅关闭与任务取消我们当前的stop()是“优雅关闭”即等待所有已入队任务完成。有时我们可能需要“立即关闭”即丢弃所有未开始执行的任务。实现立即关闭在stop()中设置标志后先清空任务队列。然后notify_all()唤醒所有线程。工作线程被唤醒后发现stop_为真且队列为空就会退出。对于被丢弃的任务其关联的future需要被设置为一个特定的异常如std::future_error否则调用future.get()的线程会永远阻塞。这需要在清空队列时对每个任务调用某种“设置异常”的操作。由于我们的任务被包装在Lambda里没有直接暴露packaged_task实现起来比较麻烦。一种方法是让PriorityTask结构体也存储std::promise或能访问future的句柄。任务取消是更复杂的功能通常需要给每个任务一个ID并提供cancel(task_id)接口。线程池需要维护一个从ID到任务的映射并在任务开始执行前检查是否被取消。这通常需要更复杂的数据结构。5.4 线程池的监控与调试对于长期运行的服务监控线程池状态很有用。可以添加以下接口size_t pending_tasks() const: 返回当前队列中等待的任务数。size_t active_threads() const: 返回正在执行任务的线程数近似值需要原子计数器。size_t total_threads() const: 返回线程池总线程数。实现active_threads可以在worker_thread中执行任务前对原子计数器加1执行完后减1。6. 常见问题、性能陷阱与排查实录即使实现了上述所有功能在实际使用中还是会遇到各种坑。下面分享一些我踩过的坑和对应的解决方案。6.1 死锁锁的粒度与顺序问题场景在worker_thread函数中如果你在持有queue_mutex_锁的情况下调用了某个用户任务而这个用户任务内部又试图向同一个线程池提交新任务submit就会发生死锁。因为submit也需要获取queue_mutex_但锁已经被当前线程持有且不会释放因为任务还没执行完。解决方案我们之前的实现已经避免了这个问题——在锁外执行任务。这是线程池设计的一条黄金法则锁只用于保护共享数据的访问执行耗时操作一定要在锁外。确保你的task()调用发生在unique_lock析构锁释放之后。6.2 惊群效应 (Thundering Herd)问题场景当有新任务入队时我们调用condition_.notify_one()。但如果有多个线程在等待操作系统可能会唤醒多个线程尽管notify_one的本意是唤醒一个。这些被唤醒的线程会同时竞争锁但只有一个能成功取出任务其他线程获取锁后发现队列又空了只能再次等待。这导致了不必要的上下文切换和锁竞争。解决方案我们的代码逻辑本身是能处理这种情况的因为每个被唤醒的线程在wait返回后都会重新检查条件stop_ || !tasks_.empty()。只有一个线程能成功通过检查并取出任务其他线程检查失败会继续等待。所以“惊群”不会导致逻辑错误只会带来一些性能损耗。在极高并发场景下可以考虑使用更复杂的通知策略但notify_one在大多数情况下已经足够好。使用notify_all()时惊群效应更明显所以只在关闭线程池等需要唤醒所有线程的场景使用。6.3 任务执行时间过长阻塞队列问题场景如果某个任务执行时间非常长如IO阻塞、死循环那么执行该任务的工作线程在此期间无法处理新任务。如果所有工作线程都被长任务占用新提交的任务就会在队列中堆积响应延迟增加。解决方案增加线程数这是最简单的方法但线程不是越多越好上下文切换开销。使用异步IO将阻塞式IO操作改为异步避免工作线程被挂起。分离线程池根据任务类型使用不同的线程池。例如CPU密集型任务一个池IO密集型任务另一个池。实现任务超时/取消但这需要操作系统或第三方库支持如使用std::jthread的中断功能或平台特定的API。6.4std::future的析构阻塞问题问题细节std::future的析构函数默认行为是如果这个future是通过std::async启动的异步任务的共享状态并且任务还未完成那么析构函数会阻塞等待任务完成。在我们的线程池中future是由std::packaged_task创建的其析构行为取决于实现。通常如果future是最后一个引用其共享状态的对象析构可能不会阻塞。但为了安全最佳实践是不要忽略std::future对象。要么调用get()或wait()明确等待结果要么将其存储起来稍后处理。避免创建了future却立刻销毁它。示例// 可能有问题future被临时创建并立即销毁 pool.submit([](){ /* 长时间任务 */ }); // 好的做法存储future或等待 auto fut pool.submit(...); // 或者明确等待如果不需要结果 pool.submit(...).wait();6.5 性能测试与线程数设置线程池的最佳线程数是多少这是一个经典问题。没有放之四海而皆准的答案但可以参考以下经验CPU密集型任务线程数最好等于或略多于CPU核心数。过多的线程会导致频繁的上下文切换降低整体吞吐量。可以使用std::thread::hardware_concurrency()获取硬件支持的并发线程数。IO密集型任务线程数可以远多于CPU核心数因为线程大部分时间在等待IO如网络、磁盘不会占用CPU。线程数可以设置到几十甚至上百具体取决于IO延迟和系统资源。混合型任务需要根据 profiling 结果来调整。一个常用的起点是线程数 CPU核心数 * (1 平均等待时间 / 平均计算时间)。测试建议实现线程池后务必进行压力测试。可以提交大量短任务观察任务完成的总时间、CPU使用率、内存占用。使用工具如perf、vtune或简单的计时器来评估性能。7. 完整代码示例与使用指南最后我将提供一个整合了返回值、异常安全、优雅关闭等特性的完整ThreadPool类并附上详细的使用说明。// thread_pool.h #pragma once #include vector #include queue #include thread #include mutex #include condition_variable #include future #include functional #include atomic #include type_traits class ThreadPool { public: explicit ThreadPool(size_t threads 0); ~ThreadPool(); ThreadPool(const ThreadPool) delete; ThreadPool operator(const ThreadPool) delete; // 提交任务返回future templateclass F, class... Args auto submit(F f, Args... args) - std::futuretypename std::invoke_result_tF, Args...; // 优雅停止等待所有任务完成 void stop(); size_t pending_tasks() const { std::lock_guardstd::mutex lock(queue_mutex_); return tasks_.size(); } size_t thread_count() const { return workers_.size(); } private: void worker_thread(); std::vectorstd::thread workers_; std::queuestd::functionvoid() tasks_; mutable std::mutex queue_mutex_; std::condition_variable condition_; std::atomicbool stop_{false}; }; inline ThreadPool::ThreadPool(size_t threads) { if (threads 0) { threads std::thread::hardware_concurrency(); if (threads 0) threads 1; } workers_.reserve(threads); for (size_t i 0; i threads; i) { workers_.emplace_back(ThreadPool::worker_thread, this); } } inline ThreadPool::~ThreadPool() { if (!stop_.load()) { stop(); } } templateclass F, class... Args auto ThreadPool::submit(F f, Args... args) - std::futuretypename std::invoke_result_tF, Args... { using return_type std::invoke_result_tF, Args...; auto task_ptr std::make_sharedstd::packaged_taskreturn_type()( std::bind(std::forwardF(f), std::forwardArgs(args)...) ); std::futurereturn_type res task_ptr-get_future(); { std::lock_guardstd::mutex lock(queue_mutex_); if (stop_.load()) { throw std::runtime_error(submit on stopped ThreadPool); } tasks_.emplace([task_ptr]() { (*task_ptr)(); }); } condition_.notify_one(); return res; } inline void ThreadPool::worker_thread() { while (true) { std::functionvoid() task; { std::unique_lockstd::mutex lock(queue_mutex_); condition_.wait(lock, [this] { return stop_.load() || !tasks_.empty(); }); if (stop_.load() tasks_.empty()) { return; } task std::move(tasks_.front()); tasks_.pop(); } task(); } } inline void ThreadPool::stop() { stop_.store(true); condition_.notify_all(); for (std::thread worker : workers_) { if (worker.joinable()) { worker.join(); } } workers_.clear(); }使用指南基本使用创建池提交任务获取结果。ThreadPool pool(4); auto future pool.submit([](int a, int b) { return a b; }, 10, 20); int result future.get(); // 阻塞直到任务完成得到30批量任务与异常处理std::vectorstd::futureint futures; for (int i 0; i 100; i) { futures.push_back(pool.submit([i]() - int { if (i % 7 0) throw std::runtime_error(Bad number); return i * i; })); } for (auto fut : futures) { try { std::cout fut.get() ; } catch (const std::exception e) { std::cout [Error] ; } }等待所有任务完成调用stop()会等待所有已提交任务完成。如果你不想停止池子只是想等待一批特定任务完成可以收集它们的future然后对每个调用wait()或get()。// 等待一批特定任务 std::vectorstd::futurevoid batch; for (...) { batch.push_back(pool.submit(...)); } for (auto f : batch) { f.wait(); } // 池子还可以继续用资源管理利用RAII让ThreadPool对象在作用域结束时自动清理。{ ThreadPool pool; // ... 使用pool ... } // 离开作用域pool.~ThreadPool()被调用自动stop和join这个线程池实现已经具备了核心功能代码清晰适合学习、面试和大多数项目需求。你可以以此为基础根据前面讨论的进阶话题添加动态线程调整、优先级队列等特性使其更加强大。记住并发编程没有银弹最好的设计总是依赖于具体的应用场景。理解原理灵活运用才是王道。