第12章线程池设计与实现

课程目录第12章线程池设计与实现

12、线程池设计与实现：线程池核心组件、任务队列设计、工作线程管理、动态调整线程数、C++11线程库实战

线程池这东西，说白了就是「复用线程」。

你想想看，如果每个请求来了都 new 一个线程，处理完再销毁——高并发下光是线程创建销毁的开销就能把 CPU 吃垮。我在早期做网关项目时就踩过这个坑，压测到 2000 QPS 时服务器直接卡死，一查发现线程数飙到了 3000 多，上下文切换占用了 80% 的 CPU 时间。嗯，从那以后，线程池就成了我工具箱里的标配。

12.1 线程池的核心组件

一个工业级的线程池，至少需要四个东西：

任务队列：存放待执行的任务，通常是 std::function<void()> 的队列
工作线程集合：一组持续运行的线程，从队列里取任务执行
同步原语：互斥锁 + 条件变量，保证线程安全
控制状态：比如是否停止、当前线程数、活跃线程数等

我个人习惯把线程池设计成「生产者-消费者」模型。主线程是生产者，往队列里塞任务；工作线程是消费者，循环取任务执行。中间用条件变量来通知——有任务时唤醒线程，没任务时让线程休眠。

  核心原则： 任务队列是共享资源，所有操作必须加锁。但锁的粒度要尽量小，只在 push/pop 时加锁，不要在任务执行期间持有锁。

12.2 任务队列设计

任务队列的设计直接影响吞吐量。我见过不少新手直接用 std::queue 加一把大锁，结果高并发下锁竞争严重，性能还不如单线程。

这里给出一个线程安全的任务队列实现：

template<typename T>
class ThreadSafeQueue {
private:
    mutable std::mutex mutex_;
    std::queue<T> queue_;
    std::condition_variable cond_;

public:
    void push(T value) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(value));
        cond_.notify_one();  // 唤醒一个等待的线程
    }

    bool try_pop(T& value) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty()) return false;
        value = std::move(queue_.front());
        queue_.pop();
        return true;
    }

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

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

这里有个细节：wait_and_pop 用了条件变量的「带谓词重载」，可以避免虚假唤醒。我曾经在生产环境遇到过因为虚假唤醒导致的空队列 pop 崩溃，排查了一整天才发现是条件变量没加谓词判断。

小技巧： 如果任务量很大，可以考虑用无锁队列（如 moodycamel::ConcurrentQueue），能减少锁竞争。但无锁队列的实现非常复杂，建议先用带锁版本，性能瓶颈明确后再优化。

12.3 工作线程管理

工作线程的核心逻辑就是一个无限循环：

void worker_thread() {
    while (!stop_) {
        std::function<void()> task;
        if (tasks_.wait_and_pop(task)) {
            task();  // 执行任务
        }
    }
}

嗯，看起来简单，但有几个坑要注意：

线程退出条件：不能只靠 stop_ 标志，还要配合条件变量通知。否则线程可能一直阻塞在 wait_and_pop 里。
异常安全：如果任务抛出异常，不能让线程挂掉。要在 task() 外面包一层 try-catch。
线程命名：调试时给线程起个名字（如 pthread_setname_np），否则 gdb 里看到一堆 thread-xxxx 根本分不清。

我曾经踩过的坑： 线程池析构时，如果还有任务在执行，直接 join 线程会导致死锁。正确的做法是：先设置 stop_ 标志，然后 notify_all 唤醒所有线程，最后再 join。这样线程会在执行完当前任务后自然退出。

12.4 动态调整线程数

固定线程数有个问题：任务少时线程空转浪费资源，任务多时线程不够导致排队。动态调整就是根据负载自动增减线程数。

我常用的策略是：

核心线程数：一直存活，即使空闲也不回收
最大线程数：允许创建的最大线程数
空闲超时：超过核心线程数的线程，空闲超过一定时间就销毁

实现思路：

void adjust_thread_count() {
    size_t task_count = tasks_.size();
    size_t thread_count = threads_.size();

    if (task_count > thread_count && thread_count < max_threads_) {
        // 任务多，线程少，且未达上限 → 创建新线程
        threads_.emplace_back([this] { worker_thread(); });
    } else if (task_count < thread_count && thread_count > core_threads_) {
        // 任务少，线程多，且超过核心数 → 通知一个线程退出
        {
            std::lock_guard<std::mutex> lock(mutex_);
            exit_flag_ = true;
        }
        cond_.notify_one();
    }
}

这个函数可以定时调用（比如每秒一次），或者在每次任务入队/出队时触发。我个人偏向于定时调用，因为实时调整的抖动太大，反而影响性能。

  经验之谈： 动态调整不是万能的。如果任务执行时间极短（微秒级），频繁创建销毁线程的开销可能比收益还大。这种情况下，固定线程数 + 无锁队列反而是更好的选择。

12.5 C++11 线程库实战

下面是一个完整的线程池实现，用到了 C++11 的 std::thread、std::mutex、std::condition_variable、std::future：

class ThreadPool {
public:
    ThreadPool(size_t threads) : stop_(false) {
        for (size_t i = 0; i < threads; ++i)
            workers_.emplace_back([this] { worker_loop(); });
    }

    template<typename F, typename... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type> 
    {
        using return_type = typename std::result_of<F(Args...)>::type;

        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );

        std::future<return_type> result = task->get_future();
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            if (stop_)
                throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks_.emplace([task]() { (*task)(); });
        }
        condition_.notify_one();
        return result;
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            stop_ = true;
        }
        condition_.notify_all();
        for (std::thread &worker : workers_)
            worker.join();
    }

private:
    void worker_loop() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(queue_mutex_);
                condition_.wait(lock, [this] { 
                    return stop_ || !tasks_.empty(); 
                });
                if (stop_ && tasks_.empty())
                    return;
                task = std::move(tasks_.front());
                tasks_.pop();
            }
            task();
        }
    }

    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex queue_mutex_;
    std::condition_variable condition_;
    bool stop_;
};

使用方式很简单：

ThreadPool pool(4);
auto result = pool.enqueue([](int a, int b) { return a + b; }, 3, 4);
std::cout << result.get() << std::endl;  // 输出 7

这里用 std::packaged_task 包装任务，返回 std::future，调用方可以异步获取结果。如果你不需要返回值，也可以直接传 std::function<void()>，省掉 packaged_task 的开销。

性能提示： 如果任务非常轻量（比如只是加个锁、写个日志），建议用批量提交 + 批量执行的方式，减少锁竞争和线程切换。我在日志库中就是这么做的，吞吐量提升了 3 倍。

12.6 线程池的整体架构

下面这张图展示了线程池的核心工作流程：

12.7 避坑指南与最佳实践

最后，总结几个我在实战中反复踩过的坑：

任务队列无界 vs 有界：无界队列可能导致内存爆炸，有界队列可能导致任务丢失。我一般用有界队列 + 拒绝策略（抛异常、丢弃、阻塞调用方）。
线程数设置：CPU 密集型任务设为 std::thread::hardware_concurrency()，IO 密集型可以设大一些（比如 2~4 倍）。没有万能公式，压测才是王道。
任务优先级：如果有些任务需要优先执行，可以用优先队列代替普通队列。但要注意优先级反转问题。
线程局部存储：如果工作线程需要维护一些状态（比如数据库连接），可以用 thread_local 变量，避免每次任务都重新创建。

我曾经犯过的错： 在析构函数中直接 join() 所有线程，但忘了先设置 stop_ 标志。结果线程池析构时，工作线程还在 wait_and_pop 里阻塞着，导致程序卡死。正确的顺序是：先设标志 → 通知所有线程 → 再 join。

线程池的设计看似简单，但要做好并不容易。从任务队列的锁粒度，到动态调整的时机，再到异常处理和资源回收，每一个细节都值得反复推敲。希望今天的分享能帮你少走一些弯路。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

12、线程池设计与实现：线程池核心组件、任务队列设计、工作线程管理、动态调整线程数、C++11线程库实战

12.1 线程池的核心组件

12.2 任务队列设计

12.3 工作线程管理

12.4 动态调整线程数

12.5 C++11 线程库实战

12.6 线程池的整体架构

12.7 避坑指南与最佳实践