AI 摘要

本文提出一种支持返回值传递且具备异常安全性的现代C++线程池实现，深入任务队列、条件变量协同机制，并剖析丢失唤醒、可拷贝包装等关键并发陷阱，为面试与工程实践提供高可用范式。

前言

在 C++ 后端开发的面试中，“手写线程池”是一道出场率极高的经典面试题。它不仅考察面试者对 std::thread、std::mutex、std::condition_variable 等并发原语的基本调用，更考验对并发陷阱（如丢失唤醒、死锁）的理解，以及对现代 C++ 特性（如智能指针、完美转发、std::future）的综合运用能力。

本文将提供一份不仅能拿满面试分数，且具有一定工程健壮性的现代 C++ 线程池实现，并深度剖析其中的核心考点。

核心设计思路

一个完备的线程池主要由以下三个部分组成：

  1. 任务队列（Task Queue）：用于存放待执行的任务。为了通用性，我们使用 std::function<void()> 来擦除不同任务参数和返回值的具体类型差异。
  2. 工作线程（Worker Threads）：在池中长期驻留的线程，它们不断尝试获取互斥锁，从队列中取出任务并执行。
  3. 同步机制（Synchronization）：使用 std::mutex 保护任务队列和退出标志，使用 std::condition_variable 来实现工作线程的休眠与唤醒，避免 CPU 忙等待。

完整代码实现

以下是支持 获取任务返回值（std::future） 且 异常安全 的完整线程池代码：

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <stdexcept>
#include <iostream>

class ThreadPool {
private:
    // 工作线程容器
    std::vector<std::thread> workers;
    // 任务队列，存储类型擦除后的 void() 可调用对象
    std::queue<std::function<void()>> qu;

    // 互斥锁，保护任务队列及 stop 标志的并发访问
    std::mutex mu;
    // 条件变量，用于工作线程等待任务或停止信号
    std::condition_variable condition;
    // 停止标志（由 mu 保护，不再需要 std::atomic）
    bool stop; 

public:
    /// @brief 创建一个包含 n 个线程的线程池
    explicit ThreadPool(size_t n) : stop(false) {
        for (size_t i = 0; i < n; ++i) {
            workers.emplace_back([this] {
                // 每个工作线程的生命周期循环
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(this->mu);
                        // 等待条件：收到停止信号 或 任务队列非空
                        this->condition.wait(lock, [this] {
                            return this->stop || !this->qu.empty();
                        });

                        // 收到退出信号且队列中已无积压任务，线程安全退出
                        if (this->stop && this->qu.empty()) {
                            return; 
                        }

                        // 从队列取出任务
                        task = std::move(this->qu.front());
                        this->qu.pop();
                    } // 释放锁，允许其他线程同时获取/提交任务

                    // 执行任务并捕获可能的异常，保证线程池不被单次任务崩溃拖垮
                    try {
                        task();
                    } catch (const std::exception& e) {
                        // 实际工程中可接入日志系统记录异常
                        std::cerr << "Task exception: " << e.what() << "\n";
                    } catch (...) {
                        std::cerr << "Unknown task exception!\n";
                    }
                }
            });
        }
    }

    /// @brief 异步提交任务，支持任意参数，并返回 std::future 以获取结果
    template<class F, class... Args>
    auto push(F&& func, Args&&... args) -> std::future<decltype(func(args...))> {
        using return_type = decltype(func(args...));

        // 核心技巧：将仅支持移动的 packaged_task 包装进 shared_ptr 中
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(func), std::forward<Args>(args)...)
        );

        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(mu);
            if (stop) {
                throw std::runtime_error("Cannot push task: thread pool is stopping.");
            }
            // Lambda 值捕获 shared_ptr，由于 shared_ptr 是可拷贝的，
            // 从而满足了 std::function 对可调用对象必须 CopyConstructible 的要求
            qu.emplace(
() { 
                (*task)(); 
            });
        } // 解锁

        // 唤醒一个可能在等待的线程
        condition.notify_one();
        return res;
    }

    /// @brief 析构函数，等待所有任务执行完毕并安全销毁线程
    ~ThreadPool() {
        {
            // 注意：必须加锁修改 stop，防止 Lost Wakeup (丢失唤醒) 导致死锁
            std::unique_lock<std::mutex> lock(mu);
            stop = true;
        }
        condition.notify_all(); // 唤醒所有挂起的线程

        for (auto& w : workers) {
            if (w.joinable()) {
                w.join();
            }
        }
    }

    // 禁用拷贝和赋值构造，防止线程池被意外复制
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
};

测试用例

我们可以编写一个简单的测试用例，验证线程池能否正确接收任务并返回结果：

int main() {
    // 创建一个包含 4 个工作线程的线程池
    ThreadPool pool(4);

    // 提交一个普通函数任务
    auto future1 = pool.push([](int a, int b) {
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        return a + b;
    }, 10, 20);

    // 提交一个会抛出异常的任务
    auto future2 = pool.push([]() -> std::string {
        throw std::runtime_error("Simulated error in task");
        return "Success";
    });

    // 获取正常任务的结果（此处会阻塞直到任务完成）
    std::cout << "Result 1: " << future1.get() << std::endl;

    // 尝试获取抛出异常的任务结果
    try {
        future2.get(); // 这里会重新抛出任务内的异常
    } catch (const std::exception& e) {
        std::cout << "Caught exception from future2: " << e.what() << std::endl;
    }

    return 0; // 退出主域时，pool 析构，优雅清理所有线程
}

面试进阶考点解析（踩坑预警）

在手撕线程池时，面试官往往会针对以下几个细节进行深度追问。理解这些细节，是区分“背诵代码”和“深刻理解并发”的试金石。

1. 为什么析构函数里的 stop = true 必须加互斥锁？

很多人喜欢把 stop 声明为 std::atomic<bool> 并在析构时直接无锁赋值。这会引发严重的 丢失唤醒（Lost Wakeup） 问题。
假设发生以下执行序列：

  1. 工作线程 拿到锁，检查 stop 发现是 false，准备调用 condition.wait() 进入休眠。此时锁即将被释放，但线程尚未真正挂起。
  2. 主线程（析构函数） 介入，无锁执行了 stop = true 和 condition.notify_all()。
  3. 工作线程 真正开始休眠。

后果：由于主线程的唤醒信号已经发完，工作线程完美错过了唤醒信号，永远沉睡。随后的 w.join() 会导致主线程永久死锁。
结论：stop 状态的修改，必须和条件变量的检查被同一把互斥锁保护。一旦受到锁的保护，stop 就不再需要声明为 std::atomic 了。

2. 为什么不能直接将 std::packaged_task 放进队列？

当我们想要返回 std::future 时，必须用到 std::packaged_task。但 std::packaged_task 是一个仅限移动（Move-Only）的类型。
然而，我们的任务队列类型是 std::queue<std::function<void()>>。C++标准库对 std::function 有硬性规定：内部存储的可调用对象必须是可拷贝构造的（CopyConstructible）。
如果直接把 packaged_task 移入 std::function，编译器会无情报错。

破局之法：我们在堆上创建一个 packaged_task，并用 std::shared_ptr 管理它。因为 shared_ptr 是可以拷贝的！我们通过 Lambda 表达式按值捕获这个 shared_ptr，巧妙地绕过了编译器的限制。

3. 工作线程为什么要加 try-catch？

线程的生命周期是非常脆弱的。如果向线程池提交的一个任务内部抛出了未捕获的异常，这个异常会直接传递到外层的工作线程函数中。一旦到达 std::thread 的最顶层，程序会调用 std::terminate() 导致整个进程直接崩溃。
加上 try-catch 后，即使个别任务引发异常（如越界、抛出自定义错误），也会被线程池消化，其他排队中的任务依然可以正常执行，保证了服务的高可用性。