手写一个线程池：任务队列、工作线程与停止顺序的三个设计点

每个任务 new 一个 std::thread 是并发编程里最贵的偷懒方式：线程创建要走系统调用，栈空间按兆算，销毁又要再来一遍。当任务本身只有几微秒时，你花在「造线程」上的时间会比干活多几十倍。 线程池的思路很朴素：线程只创建一次，任务排队，谁空谁取。这篇从零把它写出来，然后把三个最容易写错的设计点讲透。

1. 引子：算一笔线程的账

// 片段：每个任务一个线程 —— 反例，不要这么写
for (int i = 0; i < 10000; ++i) {
std::thread([i] { do_something_small(i); }).detach(); // 一万次创建 + 一万次销毁
}

这段代码有三个问题叠在一起：① 线程数不受控，任务一多直接把系统线程上限打满（std::system_error: Resource temporarily unavailable 就是这么来的）；② 创建/销毁的开销和任务量成正比，而不是和「峰值并发度」成正比；③ 拿不到返回值、接不住异常（《std::async、future、promise：异步任务的返回值、异常与三个经典坑》讲过）。

账大概是这个量级：

结论很清楚：任务越短，线程池的收益越大。三条路里，detach 是失控，逐任务建线程是浪费，线程池才是稳态。

官方文档：std::thread — cppreference · std::condition_variable — cppreference

2. 结构长什么样

┌────────────────────────────────────────────────┐
│ ThreadPool │
提交任务 ──────────► │ enqueue(task) │
（任意线程都能调） │ │ │
│ ▼ ① 加锁，把任务推进队列 │
│ ┌──────────────────────────────┐ │
│ │ tasks_ : std::queue<…> │ │
│ │ [t3][t2][t1] ← 先进先出 │ 任务队列 │
│ └──────────────────────────────┘ │
│ │ ② 解锁，再 notify_one │
│ ▼ │
│ ╔════════════════════════════════╗ │
│ ║ mutex_ + cv_ 保护队列与标志 ║ │
│ ╚════════════════════════════════╝ │
│ │ ③ 某一个正在 wait 的工作线程被唤醒 │
│ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ worker 0 │ │ worker 1 │ │ worker 2 │ … │
│ │ 循环： │ │ 循环： │ │ 循环： │ │
│ │ 等 → 取 │ │ 等 → 取 │ │ 等 → 取 │ │
│ │ → 执行 │ │ → 执行 │ │ → 执行 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└────────────────────────────────────────────────┘
↑ 线程只在构造时创建，之后被任务反复复用

四个零件缺一不可：任务队列（存放待办，std::queue 的 FIFO 语义天然公平）、工作线程数组（复用实体）、std::mutex（保护队列和停止标志）、std::condition_variable（让空闲线程真正睡着，而不是忙等烧 CPU）。

3. 最小可用版本：先跑起来

先做一版 enqueue 只收 std::function<void()> 的，把骨架跑通：

// pool_min.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread pool_min.cpp -o pool_min
#include <atomic>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <utility>
#include <vector>

class ThreadPool {
public:
explicit ThreadPool(std::size_t workers) {
workers_.reserve(workers);
for (std::size_t i = 0; i < workers; ++i) {
workers_.emplace_back([this] { worker_loop(); }); // 构造即开工
}
}

~ThreadPool() { shutdown(); }

ThreadPool(const ThreadPool&) = delete; // 线程不可拷贝，池也不该
ThreadPool& operator=(const ThreadPool&) = delete;

void enqueue(std::function<void()> task) {
{
std::lock_guard<std::mutex> lock(mutex_);
tasks_.push(std::move(task)); // ① 持锁改队列
}
cv_.notify_one(); // ② 解锁后再通知
}

private:
void shutdown() {
{
std::lock_guard<std::mutex> lock(mutex_);
stopping_ = true; // ① 先置停止标志（持锁）
}
cv_.notify_all(); // ② 再唤醒所有睡着的线程
for (auto& worker : workers_) { // ③ 最后 join
if (worker.joinable()) worker.join();
}
}

void worker_loop() {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return stopping_ || !tasks_.empty(); });
if (stopping_ && tasks_.empty()) return; // 停止且已排空 → 才能退出
task = std::move(tasks_.front());
tasks_.pop();
}
task(); // ④ 必须在锁外执行
}
}

std::mutex mutex_;
std::condition_variable cv_;
std::queue<std::function<void()>> tasks_;
std::vector<std::thread> workers_;
bool stopping_{false}; // 受 mutex_ 保护，无需 atomic
};

int main() {
constexpr int kTasks = 100;
std::atomic<int> done{0};

{
ThreadPool pool(4);
for (int i = 0; i < kTasks; ++i) {
pool.enqueue([&done] { done.fetch_add(1, std::memory_order_relaxed); });
}
} // pool 析构 → 等所有已提交任务跑完

std::printf("提交 %d 个任务, 完成 %d 个\\n", kTasks, done.load());
}

提交 100 个任务, 完成 100 个

100 == 100 是确定的，和 4 个工作线程谁先谁后无关 —— 这是析构必须把队列排空换来的保证。

4. 设计点 ①：让调用方拿到返回值

上面的 enqueue 只能收 void()，用完就是「扔出去不问结果」。标准做法是把它改成可变参数模板，内部用 std::packaged_task 包一层，返回 std::future：

// 片段：enqueue 的模板版核心
template <typename F, typename... Args>
auto enqueue(F&& func, Args&&... args)
-> std::future<std::invoke_result_t<F, Args...>> {
using Result = std::invoke_result_t<F, Args...>;

auto bound = std::make_shared<std::packaged_task<Result()>>( // ① 包起来
[func = std::forward<F>(func),
args = std::make_tuple(std::forward<Args>(args)...)]() mutable -> Result {
return std::apply(func, std::move(args)); // 参数从 tuple 里摊开
});

std::future<Result> result = bound->get_future(); // ② 先取 future
{
std::lock_guard<std::mutex> lock(mutex_);
tasks_.emplace([bound] { (*bound)(); }); // ③ 再入队
}
cv_.notify_one();
return result; // ④ 把通道交给调用方
}

这段代码里有四个必须这么写的地方，每一个都是踩出来的：

// pool_future.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread pool_future.cpp -o pool_future
#include <atomic>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <thread>
#include <tuple>
#include <type_traits>
#include <utility>
#include <vector>

class ThreadPool {
public:
explicit ThreadPool(std::size_t workers) {
workers_.reserve(workers);
for (std::size_t i = 0; i < workers; ++i) {
workers_.emplace_back([this] { worker_loop(); });
}
}

~ThreadPool() { shutdown(); }

ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;

template <typename F, typename... Args>
auto enqueue(F&& func, Args&&... args)
-> std::future<std::invoke_result_t<F, Args...>> {
using Result = std::invoke_result_t<F, Args...>;

auto bound = std::make_shared<std::packaged_task<Result()>>(
[func = std::forward<F>(func),
args = std::make_tuple(std::forward<Args>(args)...)]() mutable -> Result {
return std::apply(func, std::move(args));
});

std::future<Result> result = bound->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
tasks_.emplace([bound] { (*bound)(); });
}
cv_.notify_one();
return result;
}

private:
void shutdown() {
{
std::lock_guard<std::mutex> lock(mutex_);
stopping_ = true;
}
cv_.notify_all();
for (auto& worker : workers_) {
if (worker.joinable()) worker.join();
}
}

void worker_loop() {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return stopping_ || !tasks_.empty(); });
if (stopping_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}

std::mutex mutex_;
std::condition_variable cv_;
std::queue<std::function<void()>> tasks_;
std::vector<std::thread> workers_;
bool stopping_{false};
};

int main() {
ThreadPool pool(3);

std::vector<std::future<long long>> futures;
futures.reserve(10);
for (int i = 1; i <= 10; ++i) {
futures.push_back(pool.enqueue([](int x) { return static_cast<long long>(x) * x; }, i));
}

long long total = 0;
for (auto& fut : futures) total += fut.get(); // 每个 future 只 get 一次
std::printf("1..10 的平方和 = %lld, 期望 = %d\\n", total, 385);

// 任务内部抛出的异常，同样沿着 future 传回调用方
std::future<int> bad = pool.enqueue([]() -> int {
throw std::runtime_error("任务内部出错");
});
try {
bad.get();
} catch (const std::runtime_error& e) {
std::printf("捕获任务异常: %s\\n", e.what());
}
}

1..10 的平方和 = 385, 期望 = 385
捕获任务异常: 任务内部出错

注意最后那段：线程池不需要自己写异常处理框架，因为 packaged_task 已经把「异常存进共享状态」做掉了（机制见《std::async、future、promise：异步任务的返回值、异常与三个经典坑》）。工作线程只管 task()，异常不会漏到线程外面去触发 std::terminate。

官方文档：std::packaged_task — cppreference · std::invoke_result — cppreference

5. 设计点 ②③：停止标志与「置标志 → notify → join」的顺序

停止标志只有两个要求：读它和写它不能数据竞争。两种合法做法：

关键在析构的顺序，三步一步都不能换：

~ThreadPool() 的正确顺序与每一步的必要性

① { lock_guard lk(mutex_); stopping_ = true; } ← 置标志必须在锁内
│ 因为 wait 的谓词要读它，必须和队列状态看到同一个「快照」
│
② cv_.notify_all(); ← 叫醒所有人
│ 只置标志不通知 → 线程还睡在 wait 上，join 会永远卡住
│ （靠虚假唤醒"碰巧"醒来的那种代码不能要）
│
③ for (worker : workers_) worker.join(); ← 等所有人退出
join 必须在 notify 之后 → 线程会先在循环里把队列排空再退出
反过来先 join 再置标志 → 死锁：线程等队列，join 等线程

三种典型写错的后果：

顺带一句 C++20：标准新增了 std::jthread，它在析构时自动「请求停止 + join」（走 std::stop_token），把上面这三步做成了语言级保证 —— 需要 C++20，且它解决的是「一个线程」，池的停止逻辑还是得自己写。

官方文档：std::jthread — cppreference（C++20，析构自动 join，对比看更清楚 RAII 该做什么）

6. 设计点 ④：任务必须在锁外执行

worker_loop 里 task() 那行如果在 lock 的作用域内调用，会出两个问题：

所以标准姿势是只在临界区里「取任务」，出了花括号再执行：

临界区尽可能短：只包住「队列状态」的读写

┌── 持锁 ────────────────────┐
│ 检查谓词（stopping_ / 队列） │
│ 取走一个任务 │
└────────────────────────────┘
┌── 不持锁 ──────────────────┐
│ task() ← 耗时全在这里 │
│ 任务里可以再 enqueue │
└────────────────────────────┘

官方文档：C++ Core Guidelines · 并发章节（CP.20 「用 RAII 而不是裸 lock/unlock」、CP.43 「临界区开得越小越好」）

7. 完整示例：200 个任务 + 析构排空验证

把上面的设计合起来，跑一个带公式核对的完整程序：

// pool_sum.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread pool_sum.cpp -o pool_sum
#include <atomic>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <thread>
#include <tuple>
#include <type_traits>
#include <utility>
#include <vector>

class ThreadPool {
public:
explicit ThreadPool(std::size_t workers) {
workers_.reserve(workers);
for (std::size_t i = 0; i < workers; ++i) {
workers_.emplace_back([this] { worker_loop(); });
}
}

~ThreadPool() { shutdown(); }

ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;

template <typename F, typename... Args>
auto enqueue(F&& func, Args&&... args)
-> std::future<std::invoke_result_t<F, Args...>> {
using Result = std::invoke_result_t<F, Args...>;

auto bound = std::make_shared<std::packaged_task<Result()>>(
[func = std::forward<F>(func),
args = std::make_tuple(std::forward<Args>(args)...)]() mutable -> Result {
return std::apply(func, std::move(args));
});

std::future<Result> result = bound->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
tasks_.emplace([bound] { (*bound)(); });
}
cv_.notify_one();
return result;
}

private:
void shutdown() {
{
std::lock_guard<std::mutex> lock(mutex_);
stopping_ = true;
}
cv_.notify_all();
for (auto& worker : workers_) {
if (worker.joinable()) worker.join();
}
}

void worker_loop() {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return stopping_ || !tasks_.empty(); });
if (stopping_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}

std::mutex mutex_;
std::condition_variable cv_;
std::queue<std::function<void()>> tasks_;
std::vector<std::thread> workers_;
bool stopping_{false};
};

int main() {
constexpr int kTasks = 200;
constexpr long long kExpected = 2686700LL; // 1²+2²+…+200² = 200·201·401/6
std::atomic<int> executed{0};

ThreadPool pool(4);

std::vector<std::future<long long>> futures;
futures.reserve(kTasks);
for (int i = 1; i <= kTasks; ++i) {
futures.push_back(pool.enqueue([&executed](int x) {
executed.fetch_add(1, std::memory_order_relaxed);
return static_cast<long long>(x) * x;
}, i));
}

long long total = 0;
for (auto& fut : futures) total += fut.get(); // 主线程等所有结果，顺序无关

std::printf("任务数 = %d, 实际执行 = %d\\n", kTasks, executed.load());
std::printf("平方和 = %lld, 与公式一致 = %d\\n",
total, static_cast<int>(total == kExpected));

{ // 析构排空验证：池销毁时队列里剩的任务必须跑完，一个都不能丢
std::atomic<int> drained{0};
{
ThreadPool inner(2);
for (int i = 0; i < 50; ++i) {
inner.enqueue([&drained] { drained.fetch_add(1, std::memory_order_relaxed); });
}
} // inner 析构 → 排空 → 线程退出
std::printf("析构前提交 50 个, 析构后完成 = %d\\n", drained.load());
}
}

任务数 = 200, 实际执行 = 200
平方和 = 2686700, 与公式一致 = 1
析构前提交 50 个, 析构后完成 = 50

三个数字全部与调度顺序无关：任务一个不多一个不少（200 = 200、50 = 50，证明析构确实排空了队列），平方和等于公式值（证明每个任务的参数都正确送到了执行方）。这里刻意只打印「计数、总和、布尔核对」这类顺序无关的结果。工作线程谁先谁后、哪个线程跑了几个任务，都不影响这三行。

8. 延伸阅读

• std::thread — cppreference —— 线程创建/销毁的成本与 join/detach 的语义，理解「线程池省的是什么」
• std::condition_variable — cppreference —— wait(lock, pred) 为什么必须是谓词版，是 worker_loop 写法正确性的依据
• std::packaged_task — cppreference —— 「返回值 + 异常」通道的关键零件，注意它只可移动
• std::future — cppreference —— 调用方拿到的句柄，get() 一次性消费
• std::invoke_result — cppreference —— enqueue 返回值类型推导的出处
• std::jthread — cppreference —— C++20 的「析构自动 join」，对照它理解这篇的 shutdown() 顺序
• C++ Core Guidelines · 并发章节 —— CP.20/CP.43 等临界区与锁的规则

本知识库内的相关篇目：

• 《std::thread 入门：启动、join、detach 与生命周期》 —— std::thread 析构前必须 join 或 detach
• 《std::async、future、promise：异步任务的返回值、异常与三个经典坑》 —— std::thread 拿不到返回值、也传不出异常
• 《mutex 与 lock_guard / unique_lock：共享数据的正确加锁姿势》 —— 多个线程改同一个变量会怎样？这篇先讲清数据竞争（读-改-写被打断）为什么让结果随机

9. 一句话总结

线程池 = 一次创建、反复复用：构造时起 N 个工作线程，各自循环「等任务 → 持锁取任务 → 锁外执行」，任务队列由 mutex 保护、由 condition_variable 唤醒；enqueue 用可变参数模板 + packaged_task + future 把返回值和异常送回调用方；停止时顺序铁定是「持锁置标志 → notify_all → join」，反了就是死锁或丢任务；执行任务一定要放在锁外，否则任务里再 enqueue 会自锁、并行度也归零。