C++&thread 线程池

线程池是多线程编程中的重要概念，它通过预先创建一组线程并复用它们来执行多个任务，避免频繁创建和销毁线程的开销，提高程序性能。下面基于C++的std::thread库详细讲解线程池的实现和应用。

一、线程池的基本概念

线程池（Thread Pool） 是一种管理线程资源的模式，包含以下核心组件：

1. 线程集合：预先创建的一组线程（通常数量固定）。
2. 任务队列：存储待执行的任务（函数或可调用对象）。
3. 调度机制：将任务分配给空闲线程执行。

优点：

• 减少线程创建/销毁的开销，提升性能。
• 控制并发线程数量，避免资源耗尽。
• 简化线程管理，集中处理任务调度。

适用场景：

• 大量短小任务（如Web服务器处理请求）。
• 需要限制并发数的场景（如数据库连接池）。
• 需要异步执行但不希望手动管理线程的场景。

二、C++实现简单线程池

下面是一个基于C++11标准库的线程池实现：

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

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();
                }
            });
        }
    }
    
    // 析构函数：停止并清理所有线程
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        
        // 通知所有线程停止
        condition.notify_all();
        
        // 等待所有线程完成当前任务并退出
        for (std::thread &worker : workers) {
            worker.join();
        }
    }
    
    // 向线程池添加任务（返回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;
        
        // 包装任务为packaged_task，可获取返回值
        auto task = std::make_shared< std::packaged_task<return_type()> >(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        // 获取future用于返回
        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;
    }
    
private:
    // 工作线程集合
    std::vector<std::thread> workers;
    
    // 任务队列（存储可调用对象）
    std::queue<std::function<void()>> tasks;
    
    // 同步工具
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

三、线程池的使用示例

下面是一个使用线程池的简单示例，计算多个数的平方和：

int main() {
    // 创建包含4个线程的线程池
    ThreadPool pool(4);
    
    // 存储结果的future集合
    std::vector< std::future<int> > results;
    
    // 提交8个任务到线程池
    for (int i = 0; i < 8; ++i) {
        results.emplace_back(
            pool.enqueue([i] {
                std::cout << "处理任务 " << i << " 线程ID: " << std::this_thread::get_id() << std::endl;
                std::this_thread::sleep_for(std::chrono::seconds(1));  // 模拟耗时操作
                return i * i;  // 返回平方结果
            })
        );
    }
    
    // 收集所有任务的结果
    int sum = 0;
    for (auto && result : results) {
        sum += result.get();  // 等待并获取结果
    }
    
    std::cout << "平方和结果: " << sum << std::endl;  // 输出: 140 (0²+1²+...+7²)
    
    return 0;
}

四、线程池核心组件解析

1. 任务队列（std::queue<std::function<void()>>）

   • 存储待执行的任务，使用 std::function 包装任意可调用对象。
   • 通过 std::mutex 和 std::condition_variable 实现线程安全。

2. 工作线程（std::vector<std::thread>）

   • 线程池创建时启动固定数量的线程。
   • 每个线程循环从任务队列中取任务执行，无任务时等待。

3. 任务提交（enqueue 方法）

   • 使用 std::packaged_task 和 std::future 封装任务，支持获取返回值。
   • 使用 std::bind 和完美转发（std::forward）处理任意参数的函数。

4. 同步机制

   • std::condition_variable 实现线程的等待-通知模式。
   • std::mutex 保护任务队列的并发访问。

五、线程池的优化与扩展

1. 动态调整线程数

   • 根据系统负载和任务类型动态增减线程数量。

2. 任务优先级

   • 使用优先级队列（std::priority_queue）实现任务优先级调度。

3. 异常处理

   • 在任务执行时捕获异常，避免线程崩溃。

4. 线程亲和性

   • 将特定线程绑定到特定CPU核心，减少上下文切换开销。

5. 关闭策略

   • 提供“优雅关闭”选项，等待所有任务完成后再终止线程池。

六、注意事项

1. 线程数量选择

   • CPU密集型任务：线程数推荐设为CPU核心数（避免过多上下文切换）。
   • IO密集型任务：线程数可设为CPU核心数的数倍（因IO等待时线程可被复用）。

2. 避免死锁

   • 任务之间不应有循环依赖，否则可能导致所有线程阻塞。

3. 资源管理

   • 确保任务中使用的资源（如文件、网络连接）在线程池析构前正确释放。

七、C++标准库与第三方库

C++标准库（截至C++23）尚未提供官方线程池实现，但可使用：

• Boost.Thread：提供 thread_pool 类。
• Intel TBB：高性能并行计算库，包含线程池。
• Poco库：提供 ThreadPool 类。

如果需要生产级线程池，推荐使用成熟的第三方库，避免自行实现的复杂性和潜在问题。

通过线程池，可以更高效地管理多线程资源，提升程序性能，特别是在处理大量短小任务时优势明显。