HTLiuCL

线程池

3 min read

异步编程

#include <iostream>
#include <future>
#include <thread>
#include <chrono>
 
// 模拟耗时任务
int slow_task(int x) {
    std::this_thread::sleep_for(std::chrono::seconds(3)); // 假设耗时3秒
    return x * x;
}
 
int main() {
    std::cout << "开始任务...\n";
 
    // 异步执行 slow_task，不阻塞主线程
    std::future<int> result = std::async(std::launch::async, slow_task, 10);
 
    // 主线程可以干别的事情
    std::cout << "主线程先去干点别的活\n";
    for (int i = 0; i < 3; i++) {
        std::cout << "主线程工作中...\n";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
 
    // 等待结果（如果没算完，会在这里阻塞）
    int value = result.get();
    std::cout << "异步任务结果: " << value << "\n";
 
    return 0;
}

1. std::async

  • std::async 是 C++11 提供的一个工具，用来 异步启动任务。
  • 它会返回一个 std::future，你可以通过这个 future 在将来获取任务结果。

2. std::launch::async

  • 这是一个 策略参数，告诉 std::async 怎么启动任务。
  • std::launch 有两种常见取值：
    • std::launch::async：一定在 新线程 中异步运行.
    • std::launch::deferred：不新建线程，推迟执行，等到调用 future.get() 的时候才在当前线程执行。
  • 如果不指定，可能由实现决定（有的实现会混合使用 async/deferred）。

3. std::future<int> result

  • 这里声明了一个 future，模板参数是 int，因为 slow_task(10) 返回 int。
  • result 就是一个“票据”，用来在将来获取异步任务的结果。

单任务队列线程池

#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <atomic>
 
class ThreadPool {
public:
    ThreadPool(size_t n) : stop(false) {
        for (size_t i = 0; i < n; ++i) {
            workers.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {   // 作用域加锁
                        std::unique_lock<std::mutex> lock(this->queueMutex);
                        this->cv.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<typename F, typename... Args>
    void enqueue(F&& f, Args&&... args) {
        auto task = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            tasks.push(task);
        }
        cv.notify_one();
    }
 
    // 析构函数
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        cv.notify_all();
        for (std::thread &worker : workers)
            worker.join();
    }
 
private:
    std::vector<std::thread> workers;              // 工作线程
    std::queue<std::function<void()>> tasks;       // 任务队列
    std::mutex queueMutex;                         // 队列互斥锁
    std::condition_variable cv;                    // 条件变量
    std::atomic<bool> stop;                        // 停止标志
};

Graph View