70、STL与线程池：线程池实现、任务队列、工作窃取

多线程编程里，线程池是个老生常谈的话题。说白了，就是提前创建好一批线程，有任务来了就丢给它们干，省得频繁创建销毁线程的开销。我这些年做后端服务，几乎每个高并发项目都离不开它。

今天咱们就聊聊怎么用C++ STL的工具，手撸一个轻量级线程池。重点放在三个核心点上：任务队列怎么设计、线程怎么管理、工作窃取怎么实现。

为什么需要线程池？

你想想看，如果每个请求来了都 new 一个线程，处理完再 delete，那系统光忙活线程调度了。我曾在某个压测场景里见过，每秒上千个请求，每个请求开一个线程，结果CPU一半时间花在线程切换上，业务逻辑反而跑得慢。

线程池的好处很明显：

• 复用线程：避免频繁创建销毁
• 控制并发数：防止资源耗尽
• 任务排队：削峰填谷

核心组件一：任务队列

任务队列是线程池的“心脏”。所有待执行的任务都放在这里，工作线程从里面取任务。

我个人习惯用 std::queue 配合 std::function 来封装任务。但要注意，多线程环境下队列是共享资源，必须加锁保护。

这里我给出一个基础的任务队列封装：

#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>

class TaskQueue {
public:
    using Task = std::function<void()>;

    void push(Task task) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(task));
        cond_.notify_one();  // 通知一个等待的线程
    }

    Task pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [this] { return !queue_.empty(); });
        Task task = std::move(queue_.front());
        queue_.pop();
        return task;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }

private:
    std::queue<Task> queue_;
    mutable std::mutex mutex_;
    std::condition_variable cond_;
};

核心组件二：线程池主体

线程池管理着一组工作线程，每个线程循环从任务队列取任务执行。我习惯用 std::vector<std::thread> 来存储线程。

这里有个坑：线程池销毁时，必须优雅地通知所有线程退出。我曾经见过一个项目，析构时直接 detach 线程，结果程序退出后线程还在跑，数据全乱了。

#include <vector>
#include <thread>
#include <atomic>

class ThreadPool {
public:
    explicit ThreadPool(size_t threadCount) : stop_(false) {
        for (size_t i = 0; i < threadCount; ++i) {
            workers_.emplace_back([this] {
                while (true) {
                    Task task = tasks_.pop();  // 阻塞等待任务
                    if (stop_ && tasks_.empty()) {
                        break;  // 收到停止信号且无任务，退出
                    }
                    task();  // 执行任务
                }
            });
        }
    }

    ~ThreadPool() {
        stop_ = true;
        // 往队列里塞空任务，唤醒所有线程
        for (size_t i = 0; i < workers_.size(); ++i) {
            tasks_.push([]{});  // 空任务，让线程检查 stop_ 标志
        }
        for (auto& t : workers_) {
            if (t.joinable()) t.join();
        }
    }

    template<typename F>
    void enqueue(F task) {
        tasks_.push(std::move(task));
    }

private:
    std::vector<std::thread> workers_;
    TaskQueue tasks_;
    std::atomic<bool> stop_;
};

核心组件三：工作窃取

工作窃取（Work Stealing）是提升线程池效率的利器。基本思想是：每个线程有自己的任务队列，当自己的队列空了，就去“偷”别的线程的任务。

为什么要这么做？因为任务分配不均匀。比如线程A处理一个耗时任务，线程B处理一个轻量任务，B很快就空闲了，而A还在忙。工作窃取能让空闲线程帮忙，提高整体吞吐量。

我曾在图像处理项目里用过这个模式。每个线程处理一块图像区域，有的区域纹理复杂，有的简单。没有工作窃取时，总有几个线程闲着，整体速度上不去。加上窃取后，快了将近30%。

下面是一个简化版的工作窃取线程池：

class WorkStealingPool {
public:
    explicit WorkStealingPool(size_t threadCount) 
        : queues_(threadCount), stop_(false) {
        for (size_t i = 0; i < threadCount; ++i) {
            workers_.emplace_back([this, i] {
                while (!stop_) {
                    Task task;
                    // 先从自己的队列取
                    if (queues_[i].tryPop(task)) {
                        task();
                        continue;
                    }
                    // 自己的队列空了，随机偷别人的
                    for (size_t j = 0; j < queues_.size(); ++j) {
                        size_t victim = (i + j + 1) % queues_.size();
                        if (queues_[victim].trySteal(task)) {
                            task();
                            break;
                        }
                    }
                    // 实在没任务，休眠一会儿
                    std::this_thread::yield();
                }
            });
        }
    }

    void enqueue(size_t threadId, Task task) {
        queues_[threadId].push(std::move(task));
    }

private:
    std::vector<std::thread> workers_;
    std::vector<LockFreeQueue<Task>> queues_;  // 无锁队列
    std::atomic<bool> stop_;
};

知识体系结构图

下面这张图展示了线程池的核心组件和它们之间的关系：

避坑指南

我这些年踩过的坑，总结成几条：

• 任务不要太大：如果单个任务执行时间过长，其他任务会饿死。建议拆分成小任务，或者用异步回调。
• 注意死锁：如果任务A等待任务B的结果，而B在队列里等A执行完，就死锁了。我遇到过这种场景，排查了一下午才发现。
• 线程数不是越多越好：一般设置为 std::thread::hardware_concurrency() 即可。太多线程反而增加上下文切换开销。
• 异常安全：任务执行中抛异常怎么办？我习惯在任务外层包一层 try-catch，记录日志后继续执行，不能让一个坏任务拖垮整个池子。

性能对比

方案	适用场景	优点	缺点
单队列线程池	任务均匀、量不大	实现简单	锁竞争严重
多队列+工作窃取	任务不均匀、高并发	负载均衡好	实现复杂
无锁队列线程池	极致性能要求	无锁等待	ABA问题、调试困难

说实话，大部分场景用单队列就够了。只有当你压测发现锁竞争成为瓶颈时，才值得上工作窃取。不要过度设计。

总结

线程池的核心就三件事：任务怎么存、线程怎么取、空闲时怎么办。STL 提供了 std::thread、std::mutex、std::condition_variable 这些基础工具，组合起来就能造出好用的线程池。

工作窃取是个进阶话题，但理解它的思想比实现更重要——说白了就是“别闲着，去帮帮别人”。这种思想在很多分布式系统里也能看到。

嗯，今天就聊到这里。代码我已经放在项目里了，你可以直接拿去用。记得根据实际场景调整线程数和队列策略。