32、C++项目实战：线程池实现

线程池这东西，说白了就是「提前招一批工人，有活就干，没活待命」。

我刚开始做后端服务时，遇到一个痛点：每次来一个请求就 new 一个线程，请求处理完再销毁。高并发下，光线程创建销毁的开销就能把 CPU 吃满。后来我学乖了——用线程池。

为什么需要线程池？

你想想看，线程的创建和销毁不是免费的。每次 pthread_create 都要分配栈空间、注册信号处理、初始化 TLS……一套下来少说几十微秒。如果任务本身只跑几百微秒，那线程管理的开销占比就太大了。

线程池的核心思想就三个字：复用。把线程的生命周期和任务的生命周期解耦。

核心设计思路

一个线程池，内部其实就两样东西：一组工作线程 + 一个任务队列。

工作线程循环从队列里取任务执行。如果队列为空，线程就阻塞等待。有新任务提交时，唤醒一个空闲线程去处理。

嗯，这里要注意：任务队列是共享资源，多个线程同时操作它，必须加锁。我习惯用 std::mutex 配合 std::condition_variable 来实现生产-消费模型。

整体架构图

先看一张图，把线程池的脉络理清楚：

代码实现

下面给出一个精简但完整的 C++11 线程池实现。我尽量去掉无关细节，保留核心骨架。

#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>

class ThreadPool {
public:
    ThreadPool(size_t threads) : stop(false) {
        for (size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock, [this] {
                            return this->stop || !this->tasks.empty();
                        });
                        if (this->stop && this->tasks.empty())
                            return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    template<class F, class... 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> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            if (stop)
                throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks.emplace([task](){ (*task)(); });
        }
        condition.notify_one();
        return res;
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for (std::thread &worker : workers)
            worker.join();
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

关键点解析

1. 工作线程的循环

每个线程内部是一个 while(true) 循环。它先获取锁，然后检查条件：要么 stop 为 true（线程池要销毁），要么队列非空（有任务可做）。

这里用 condition.wait(lock, predicate) 的写法，比手动 if + wait 更安全——它能防止虚假唤醒。

2. 任务提交与 future

enqueue 返回一个 std::future，调用方可以通过它拿到任务的返回值或异常。内部用 std::packaged_task 把可调用对象包装起来，再塞进队列。

注意这里用 std::make_shared 管理 packaged_task 的生命周期——因为任务可能在线程池内部执行，而 future 在外部等待，两者必须共享同一个状态。

3. 析构时的优雅退出

析构函数先设置 stop 标志，然后 notify_all 唤醒所有线程。每个线程被唤醒后检查 stop && 队列空，满足则退出循环并结束线程。

使用示例

#include <iostream>

int main() {
    ThreadPool pool(4);

    // 提交多个任务
    auto result1 = pool.enqueue([](int a, int b) {
        return a + b;
    }, 10, 20);

    auto result2 = pool.enqueue([](int a, int b) {
        return a * b;
    }, 5, 6);

    std::cout << "10 + 20 = " << result1.get() << std::endl;
    std::cout << "5 * 6 = " << result2.get() << std::endl;

    return 0;
}

线程数怎么定？

这个问题没有标准答案。我一般按场景分：

任务类型	推荐线程数	说明
CPU 密集型	std::thread::hardware_concurrency()	线程数 ≈ CPU 核心数，避免上下文切换开销
IO 密集型	核心数 × 2 ~ 核心数 × 4	IO 等待时让出 CPU，可以多开一些线程
混合型	核心数 × (1 + 等待时间/计算时间)	需要 profiling 后调整

扩展思考

上面的实现是最基础的版本。实际项目中还可以加：

• 任务优先级队列：用 std::priority_queue 替代 std::queue
• 动态调整线程数：根据负载增减工作线程
• 任务窃取：每个线程有自己的任务队列，空闲时去偷别人的任务
• 超时机制：任务执行超时自动取消或重试

嗯，这些属于进阶话题了。先把基础版本跑通，再根据需求逐步迭代。线程池这东西，够用就好，别过度设计。

---

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