C++11实现线程池-开发者社区

这是一个线程池实现，结合了多个异步编程示例。让我详细解释每个部分：

整体结构

代码展示了多种 C++11 异步编程技术的用法，最后实现了一个完整的线程池。

各部分详解

1.第一部分：std::async 的基本用法

void task(const char* name) { std::cout << name << " 开始执行，线程ID: " << std::this_thread::get_id() << std::endl; std::this_thread::sleep_for(std::chrono::seconds(1)); std::cout << name << " 完成" << std::endl; }

演示了std::async的两种启动策略：
- std::launch::async: 立即在新线程中执行
- std::launch::deferred: 延迟执行（调用get()时才执行）

2.第二部分：std::future 获取结果

std::future<int> result = std::async(std::launch::async, simpleTask); int value = result.get();

展示了如何通过std::future::get()获取异步任务结果
注意：get()只能调用一次

3.第三部分：std::promise 和 std::future

std::promise<double> prom; auto fut = prom.get_future(); std::thread t(computeSqrt, std::move(prom), 9);

std::promise用于在线程间传递结果
可以从 promise 获取 future
可以在另一个线程中设置值或异常

4.第四部分：std::shared_future

std::shared_future<int> sharedFut = prom.get_future().share();

多个线程可以共享同一个 future
get()可以多次调用
通过share()从普通 future 创建 shared_future

5.第五部分：std::packaged_task

std::packaged_task<int(int)> task([](int x) { return x * x; }); std::future<int> result = task.get_future(); std::thread worker(std::move(task), 5);

包装可调用对象
可以获取关联的 future
可以传递给线程执行

6.第六部分：任务队列和条件变量

void taskRunner(std::queue<std::packaged_task<int()>>& taskqueue, ...) { while (true) { std::unique_lock<std::mutex> lock(task_mtx); cv.wait(lock, [&taskqueue] { return !taskqueue.empty(); }); // 处理任务 } }

使用std::condition_variable实现线程间通信
工作线程等待任务，有任务时被唤醒
主线程添加任务到队列

7.第七部分：并行计算示例（计算素数）

futures.push_back( std::async(std::launch::async, countPrimes, start, end) ); // 收集结果 int total1 = 0; for (auto& fut : futures) { total1 += fut.get(); }

将大任务拆分成多个子任务
使用std::async并行计算
收集所有结果并汇总

8.第八部分：完整的 ThreadPool 类

这是最重要的部分，实现了一个生产级别的线程池：

核心设计

class ThreadPool { private: std::vector<std::thread> workers; // 工作线程 std::queue<std::function<void()>> tasks; // 任务队列 std::mutex queue_mutex; // 队列互斥锁 std::condition_variable condition; // 条件变量 bool stop; // 停止标志 };

关键方法

构造函数：创建指定数量的工作线程
enqueue 方法：提交任务到线程池
析构函数：优雅关闭线程池

模板方法 enqueue

template<class F, class... Args> auto enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> { // 1. 推导返回类型 using return_type = typename std::result_of<F(Args...)>::type; // 2. 创建 packaged_task auto task = std::make_shared<std::packaged_task<return_type()>>( std::bind(std::forward<F>(f), std::forward<Args>(args)...) ); // 3. 获取 future std::future<return_type> res = task->get_future(); // 4. 将任务添加到队列 { std::unique_lock<std::mutex> lock(queue_mutex); tasks.emplace([task]() { (*task)(); }); } // 5. 通知一个等待的线程 condition.notify_one(); return res; }

工作线程循环

for (;;) { 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(); // 执行任务 }

9.主函数测试

int main() { const int MAX = 100000000; const int THREADS = 8; ThreadPool pool(8); // 创建线程池 std::vector<std::future<int>> futures; // 拆分任务并提交到线程池 for (int i = 0; i < THREADS; ++i) { int start = i * chunk + 1; int end = (i == THREADS - 1) ? MAX : (i + 1) * chunk; futures.push_back( pool.enqueue(countPrimes, start, end) // 提交任务 ); } // 收集结果 int total1 = 0; for (auto& fut : futures) { total1 += fut.get(); // 阻塞等待结果 } std::cout << "素数总数: " << total1 << std::endl; return 0; }