C++ 线程池原理与实现：从零构建一个可复用的线程池

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

class ThreadPool {
public:
    // 默认线程数 = 硬件并发数（CPU 核心数）
    explicit ThreadPool(size_t threads = std::thread::hardware_concurrency())
        : stop_(false) {
        for (size_t i = 0; i < threads; ++i) {
            workers_.emplace_back([this] {
                for (;;) {
                    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::future 以便获取执行结果
    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 (auto& 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_;                                    // 停止标志
};

#include <iostream>
#include <chrono>

int main() {
    ThreadPool pool(4);

    // 提交无返回值的任务
    pool.enqueue([] {
        std::cout << "task in thread " << std::this_thread::get_id() << "\n";
    });

    // 提交带参数、带返回值的任务
    auto fut = pool.enqueue([](int a, int b) { return a + b; }, 20, 22);
    std::cout << "20 + 22 = " << fut.get() << "\n";

    // 批量提交
    std::vector<std::future<int>> results;
    for (int i = 0; i < 8; ++i)
        results.emplace_back(pool.enqueue([i] { return i * i; }));
    for (auto& r : results)
        std::cout << r.get() << " ";
    std::cout << "\n";
    return 0;
}

task in thread 10324
20 + 22 = 42
0 1 4 9 16 25 36 49