手写线程池

手写线程池

组合 jthread、有界阻塞队列、packaged_task 与 future，实现可回收异常和优雅关闭的线程池。

本系列代码使用 C++20 和 oc::handmade 命名空间，目标是解释实现机制、复杂度和工程边界，不是替代标准库。普通容器与缓存核心不内置互斥锁；这不代表 lock-free。

---

一、学习目标

• 实现类型安全 submit
• 让任务异常进入 future
• 保证优雅关闭幂等且不丢已接收任务

---

二、前置条件

完成有界阻塞队列篇，熟悉 invoke、future 和可调用对象。

Linux/macOS：

cmake -S . -B build -DCMAKE_BUILD_TYPE=Debug
cmake --build build -j
ctest --test-dir build --output-on-failure

Windows PowerShell：

cmake -S . -B build -DCMAKE_BUILD_TYPE=Debug
cmake --build build --config Debug
ctest --test-dir build -C Debug --output-on-failure

---

三、问题与设计选择

工作项擦除为可移动的 packaged_task<void()>；submit 创建结果任务并返回 future。shutdown 关闭队列，工作线程排空后由 jthread 自动 join。

这里刻意保留一条边界：教学实现覆盖构造、复制移动、核心修改、查找和迭代契约，但不复刻标准库全部重载、ABI、constexpr、异构查找或节点句柄。

---

四、内存布局与核心不变量

每个成功提交任务恰执行一次；停止后提交必失败；shutdown 返回时所有工作线程已结束。

每个修改操作都按“准备资源 → 构造新状态 → 提交连接或指针 → 清理旧状态”的顺序设计。提交点之前发生异常，应保持原对象可继续使用；无法提供强保证时，会在接口说明中明确基本保证。

---

五、核心实现

template<class F, class... Args>
auto submit(F&& f, Args&&... args)
    -> std::future<std::invoke_result_t<F, Args...>> {
    using result = std::invoke_result_t<F, Args...>;
    std::packaged_task<result()> task(
        std::bind_front(std::forward<F>(f),
                        std::forward<Args>(args)...));
    auto future = task.get_future();
    if (!tasks_.push(task_type{
            [task = std::move(task)]() mutable { task(); }}))
        throw std::runtime_error("thread_pool stopped");
    return future;
}

上面先聚焦最容易写错的核心步骤；若本篇对应一个独立组件，下一节给出统一工程中的完整教学实现。代码没有放入 std 命名空间，避免未定义行为和名称冲突。

---

六、完整教学实现

下面是统一工程中经过 GCC、Clang、GoogleTest 和 Sanitizer 验证的完整组件。它依赖前序文章已经实现的公共类型以及头文件中的标准库 #include。

namespace oc::handmade {

class thread_pool {
    using task_type = std::packaged_task<void()>;
    bounded_blocking_queue<task_type> tasks_;
    std::vector<std::jthread> workers_;
    std::once_flag shutdown_once_;

    void work() {
        while (auto task = tasks_.pop()) (*task)();
    }
public:
    explicit thread_pool(
        std::size_t threads = std::max(1U, std::thread::hardware_concurrency()),
        std::size_t queue_capacity = 1024
    ) : tasks_(queue_capacity) {
        if (threads == 0) throw std::invalid_argument("thread count must be positive");
        workers_.reserve(threads);
        for (std::size_t i = 0; i < threads; ++i)
            workers_.emplace_back([this] { work(); });
    }
    thread_pool(const thread_pool&) = delete;
    thread_pool& operator=(const thread_pool&) = delete;
    ~thread_pool() { shutdown(); }
    template<class F, class... Args>
    auto submit(F&& callable, Args&&... args)
        -> std::future<std::invoke_result_t<F, Args...>> {
        using result_type = std::invoke_result_t<F, Args...>;
        std::packaged_task<result_type()> result_task(
            std::bind_front(
                std::forward<F>(callable),
                std::forward<Args>(args)...
            )
        );
        auto result = result_task.get_future();
        task_type erased(
            [task = std::move(result_task)]() mutable { task(); }
        );
        if (!tasks_.push(std::move(erased)))
            throw std::runtime_error("thread_pool is stopped");
        return result;
    }
    void shutdown() noexcept {
        std::call_once(shutdown_once_, [this] {
            tasks_.close();
            workers_.clear();
        });
    }
};

}  // namespace oc::handmade

生产级标准库还要处理完整 allocator 传播、全部重载、ABI、调试迭代器和平台特化；这里保留的是能够独立推导核心数据结构的教学边界。

---

七、使用示例与输出

预期输出或状态：

提交平方任务得到 future 49；抛异常任务在 future.get() 时重新抛出；shutdown 后 submit 被拒绝。

示例必须在文章对应的测试目标中实际编译。涉及顺序的输出只依赖接口明确承诺的顺序；无序容器不会把某次桶顺序写成稳定结果。

---

八、复杂度与失效规则

操作	复杂度	说明
submit	O(1)+可能等待	受队列背压
执行任务	取决于任务	锁外执行
shutdown	O(剩余任务)	幂等排空
析构	同 shutdown	不抛异常

复杂度中的 O(1) 若标记为“平均”或“摊还”，不能在面试中省略限定词。任何重新分配、节点删除、rehash 或缓存淘汰都必须单独说明迭代器、引用与指针是否失效。

---

九、异常安全与资源管理

• 获取资源后立即交给 RAII 对象或明确记录已构造数量。
• 用户类型构造、复制、移动、比较器和哈希器都可能抛异常。
• 只有在所有后续步骤不会失败时才修改不可回滚的链接。
• 析构、释放和关闭路径不得抛异常。
• 并发包装通过回调在锁内访问，避免返回保护对象的裸引用。

---

十、常见错误

1. 把 move-only packaged_task 放入 std::function

把 move-only packaged_task 放入 std::function会破坏本篇建立的契约。调试时先检查核心不变量，再缩小到触发该状态的最短操作序列。

2. 持队列锁执行任务

持队列锁执行任务会破坏本篇建立的契约。调试时先检查核心不变量，再缩小到触发该状态的最短操作序列。

3. 析构时直接丢弃尚未执行的 future

析构时直接丢弃尚未执行的 future会破坏本篇建立的契约。调试时先检查核心不变量，再缩小到触发该状态的最短操作序列。

---

十一、面试追问

1. 为什么选择 jthread 而不是 thread？
2. 工作窃取解决什么问题？
3. 线程数为何不应机械等于 hardware_concurrency？

回答时先说数据结构不变量，再给复杂度，最后说明异常、迭代器或并发边界，通常比背诵结论更有说服力。

---

十二、练习与自测

1. 增加 submit 超时
2. 加入任务优先级
3. 测量不同队列容量的吞吐与尾延迟

自测标准：能够不看代码画出内存或节点关系，解释一次成功操作和一次失败回滚，并写出至少一个会击穿错误实现的测试。

---

十三、官方资料与延伸阅读

• C++20 工作草案 N4861
• C++ Core Guidelines
• P0660R9 stop_token 与 jthread

---

上一篇：手写有界阻塞队列 ｜ 下一篇：手写 FIFO 缓存