当前位置：首页 > news >正文

c++线程池的实现

news 来源：原创 2025/6/17 14:26:37

一、问题引出

二、优化建议与改进方案

1. 基础加固

2. 功能增强

3. 性能优化

三、完整优化代码示例

四、关键优化点总结

1.安全性增强：

2.功能扩展：

3.性能优化：

4.工业级特性：

五、测试验证建议

一、问题引出

在阿Q技术站看到一个c++线程池的实现。该线程池还有可以改进的地方。

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

class ThreadPool{
public:
    ThreadPool(size_t num_threads) : stop(false) {
        for(size_t i = 0; i < num_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>
    void enqueue(F&& f) {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            tasks.emplace(std::forward<F>(f));
        }
        condition.notify_one();
    }

    ~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;
};

int main() {
    ThreadPool pool(4); // 创建4个线程的线程池
    for (int i = 0; i < 8; ++i) {
        pool.enqueue([i]() {
            std::cout << "Task " << i << " executed by thread " << std::this_thread::get_id() << "\n";
            std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟工作
        });
    }
    std::this_thread::sleep_for(std::chrono::seconds(1)); // 等待任务完成
    return 0;
}

逻辑问题：

在enqueue时，如果线程池已经停止（stop=true），仍然可以添加任务，这可能导致在析构后执行任务，引发未定义行为。需要在enqueue时检查stop标志。
异常处理：如果task()抛出异常，线程将终止，导致线程池中的线程数减少，影响后续任务执行。
stop标志可以改用原子类型。

二、优化建议与改进方案

1. 基础加固

问题修复代码：

// 修改 enqueue 函数
template<class F>
bool enqueue(F&& f) {  // 返回是否提交成功
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(stop) return false;  // 池已关闭拒绝任务
        tasks.emplace(std::forward<F>(f));
    }
    condition.notify_one();
    return true;
}

2. 功能增强

(1) 支持任务返回值

template<class F>
auto enqueue(F&& f) -> std::future<decltype(f())> {
    using RetType = decltype(f());
    
    auto task = std::make_shared<std::packaged_task<RetType()>>(
        std::forward<F>(f)
    );
    
    std::future<RetType> res = task->get_future();
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(stop) throw std::runtime_error("enqueue on stopped pool");
        tasks.emplace([task]() { (*task)(); });
    }
    condition.notify_one();
    return res;
}

(2) 异常安全处理

// 修改工作线程循环
try {
    task();
} catch (...) {
    // 记录异常或传递异常
    std::lock_guard<std::mutex> lock(exception_mutex);
    exceptions.push_back(std::current_exception());
}