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

C++学习：六个月从基础到就业——多线程编程：线程池实现

news 来源：原创 2025/5/23 14:17:33

C++学习：六个月从基础到就业——多线程编程：线程池实现

本文是我C++学习之旅系列的第五十九篇技术文章，也是第四阶段"并发与高级主题"的第六篇，介绍如何设计和实现一个高效、可靠的C++线程池。查看完整系列目录了解更多内容。

引言

在前面的文章中，我们已经学习了多线程编程的基础知识，包括线程创建、同步机制和异步任务等。当应用程序需要处理大量并发任务时，频繁地创建和销毁线程会带来严重的性能开销。线程池通过预先创建并管理一组工作线程来解决这个问题，从而提高任务处理效率、优化资源使用，是高性能并发应用的关键组件。

本文将系统地介绍线程池的概念、设计考量以及在C++中的实现方法，并通过一个完整的线程池实现案例，帮助你掌握这一重要的并发编程技术。

多线程编程：线程池实现
- 引言
- 目录
- 线程池概述
  - 为什么需要线程池
  - 线程池的基本组成
  - 常见的线程池模型
- 设计考量
  - 线程数量策略
  - 任务队列设计
  - 线程池生命周期
  - 负载均衡策略
  - 异常处理机制
- 基本线程池实现
  - 线程池核心结构
  - 线程池初始化
  - 任务提交与执行
  - 结果获取机制
  - 优雅关闭
- 高级特性实现
  - 动态调整线程数
  - 任务优先级支持
  - 任务取消机制
  - 线程池状态监控
- 性能优化
  - 锁竞争优化
  - 缓存友好设计
  - 避免虚假共享
  - 任务批处理
- 线程池应用场景
  - Web服务器
  - 并行计算
  - 图像处理
  - 异步IO处理
- 实际案例：高性能线程池
  - 完整代码与解析
  - 使用示例
  - 性能测试
- 注意事项与最佳实践
- 总结

线程池概述

为什么需要线程池

线程的创建和销毁是昂贵的操作，涉及到系统调用、内存分配和上下文切换等开销。频繁地创建和销毁线程会对应用程序性能产生显著影响，尤其在处理大量短期任务时。线程池通过以下方式解决这个问题：

线程复用：预先创建线程并重复使用，避免频繁创建和销毁的开销
控制并发度：限制并发线程数，防止系统资源耗尽
任务管理：提供任务队列和调度机制，平衡工作负载
异步处理：将任务提交与执行分离，提高响应性

下面的示例展示了使用线程池与直接创建线程的区别：

#include <iostream>
#include <thread>
#include <chrono>
#include <vector>// 直接创建线程的方式
void without_thread_pool(int num_tasks) {auto start = std::chrono::high_resolution_clock::now();std::vector<std::thread> threads;for (int i = 0; i < num_tasks; ++i) {threads.emplace_back([]() {// 模拟任务执行std::this_thread::sleep_for(std::chrono::milliseconds(10));});}for (auto& t : threads) {t.join();}auto end = std::chrono::high_resolution_clock::now();std::chrono::duration<double, std::milli> duration = end - start;std::cout << "Without thread pool: " << duration.count() << " ms" << std::endl;
}// 使用线程池结果示意（实际代码稍后实现）
void with_thread_pool(int num_tasks) {// 这里只是示意，我们会在后面实现真正的线程池auto start = std::chrono::high_resolution_clock::now();// ThreadPool pool(4);  // 假设使用4个工作线程// for (int i = 0; i < num_tasks; ++i) {//     pool.enqueue([]() {//         std::this_thread::sleep_for(std::chrono::milliseconds(10));//     });// }// 模拟线程池执行时间std::this_thread::sleep_for(std::chrono::milliseconds(num_tasks * 10 / 4));auto end = std::chrono::high_resolution_clock::now();std::chrono::duration<double, std::milli> duration = end - start;std::cout << "With thread pool: " << duration.count() << " ms (simulated)" << std::endl;
}int main() {constexpr int NUM_TASKS = 100;without_thread_pool(NUM_TASKS);with_thread_pool(NUM_TASKS);return 0;
}

线程池的基本组成

一个标准的线程池通常包含以下核心组件：

工作线程组：一组预先创建的线程，等待并执行提交的任务
任务队列：存储待执行的任务
同步机制：确保任务队列的线程安全访问
线程管理器：负责监控和管理工作线程
任务调度器：负责从队列中取出任务并分配给工作线程

线程池基本架构

常见的线程池模型

线程池有多种实现模型，每种模型适用于不同的应用场景：

固定大小线程池：创建固定数量的工作线程，简单可靠
动态线程池：根据负载动态调整工作线程数量
缓存线程池：按需创建线程，长时间空闲的线程会被终止
单线程池：只有一个工作线程，确保任务按顺序执行
调度线程池：支持定时或周期性执行任务

本文将主要关注固定大小线程池的实现，并在此基础上讨论更高级的特性。

设计考量

线程数量策略

选择合适的线程数量对线程池性能至关重要。线程过少无法充分利用系统资源，线程过多则会增加上下文切换开销。以下是几种常见的线程数量策略：

基于CPU核心数：最常用的策略，通常设置为CPU逻辑核心数或其倍数
```
unsigned int thread_count = std::thread::hardware_concurrency();
```
基于任务类型：根据任务是CPU密集型还是IO密集型调整线程数
- CPU密集型：线程数 = CPU核心数
- IO密集型：线程数 = CPU核心数 * (1 + 等待时间/计算时间)
自适应策略：根据系统负载和任务队列长度动态调整线程数
手动配置：允许用户根据特定需求设置线程数

任务队列设计

任务队列是线程池的核心组件，需要考虑以下几个方面：

队列类型：可以使用标准容器如std::queue，也可以使用无锁队列提高性能
队列容量：
- 无界队列：理论上可以接受无限任务，但可能导致内存问题
- 有界队列：当队列满时，可以阻塞、丢弃任务或执行其他策略
任务优先级：支持任务优先级可以确保重要任务优先执行
公平性：确保任务不会因优先级低而永远无法执行（饥饿问题）

线程池生命周期

线程池的生命周期管理包括创建、运行和关闭三个阶段：

创建阶段：
- 初始化工作线程
- 设置线程池参数
- 准备任务队列
运行阶段：
- 接收并执行任务
- 处理空闲线程
- 监控线程状态
关闭阶段：
- 平滑关闭：等待所有已提交任务完成
- 立即关闭：取消未执行的任务，立即终止
- 资源清理：释放线程资源

负载均衡策略

为了提高线程池效率，可以采用不同的负载均衡策略：

工作窃取（Work Stealing）：空闲线程从其他线程的队列"窃取"任务
轮询分配（Round Robin）：任务轮流分配给各个线程
最少任务优先：任务分配给当前负载最轻的线程
亲和性调度：考虑CPU缓存亲和性，尽量让相关任务在同一核心上执行

异常处理机制

线程池需要妥善处理任务执行过程中可能出现的异常：

任务内部异常捕获：在任务内部捕获并处理异常，避免影响线程池
线程级异常处理：在工作线程中捕获任务异常，确保线程不会意外终止
异常传递机制：将任务异常传递给调用者（通过std::future等机制）
异常回调：提供回调机制，通知调用者任务执行过程中的异常

基本线程池实现

下面，我们将实现一个基本的线程池，功能包括任务提交、执行和结果获取。

线程池核心结构

首先，我们定义线程池的核心类结构：

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <functional>
#include <vector>
#include <future>
#include <memory>
#include <stdexcept>
#include <type_traits>class ThreadPool {
private:// 工作线程std::vector<std::thread> workers;// 任务队列std::queue<std::function<void()>> tasks;// 同步相关std::mutex queue_mutex;std::condition_variable condition;bool stop;public:// 构造函数：创建指定数量的工作线程explicit ThreadPool(size_t threads);// 提交任务到线程池，返回std::futuretemplate<class F, class... Args>auto enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type>;// 析构函数：停止所有线程~ThreadPool();
};

线程池初始化

线程池的初始化是创建工作线程并设置它们的工作循环：

ThreadPool::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(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 ThreadPool::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// std::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::lock_guard<std::mutex> lock(queue_mutex);// 线程池已停止，不能再添加任务if (stop) {throw std::runtime_error("enqueue on stopped ThreadPool");}// 将任务添加到队列tasks.emplace([task]() { (*task)(); });}// 通知一个等待中的线程有新任务condition.notify_one();return res;
}

结果获取机制

通过std::future，我们可以异步获取任务的执行结果：

// 示例：使用线程池并获取结果
int calculate(int a, int b) {return a + b;
}void result_example() {ThreadPool pool(4);// 提交任务并获取futureauto result = pool.enqueue(calculate, 10, 20);// 获取结果（如果任务尚未完成，会阻塞等待）int sum = result.get();std::cout << "Result: " << sum << std::endl;
}

优雅关闭

线程池的析构函数负责安全地关闭所有工作线程：

ThreadPool::~ThreadPool() {{// 设置stop标志std::lock_guard<std::mutex> lock(queue_mutex);stop = true;}// 通知所有等待中的线程condition.notify_all();// 等待所有工作线程结束for (std::thread& worker : workers) {worker.join();}
}

高级特性实现

基本线程池已经可以满足大多数需求，但在特定场景下，我们可能需要一些高级特性。

动态调整线程数

实现动态调整线程数量的功能，可以根据负载情况增加或减少工作线程：

class DynamicThreadPool {// ...基本线程池成员...size_t min_threads;size_t max_threads;std::atomic<size_t> active_threads; // 当前活跃线程数public:// 根据负载调整线程数void adjust_thread_count() {std::lock_guard<std::mutex> lock(queue_mutex);size_t current = workers.size();size_t active = active_threads.load();size_t pending = tasks.size();// 负载过高，增加线程if (current < max_threads && active == current && pending > current) {// 添加新线程add_worker();}// 负载过低，减少线程else if (current > min_threads && active < current / 2 && pending == 0) {// 减少一个线程remove_worker();}}private:void add_worker() {// 创建新工作线程...}void remove_worker() {// 移除一个工作线程...}
};

任务优先级支持

为任务添加优先级支持，确保高优先级任务优先执行：

class PriorityThreadPool {
private:// 带优先级的任务struct PriorityTask {int priority;std::function<void()> task;// 比较运算符重载，用于优先队列bool operator<(const PriorityTask& other) const {return priority < other.priority;}};// 使用优先队列替代普通队列std::priority_queue<PriorityTask> tasks;// ...其他线程池成员...public:// 提交带优先级的任务template<class F, class... Args>auto enqueue_with_priority(int priority, F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> {// ...类似于基本enqueue，但添加优先级...}
};

任务取消机制

实现任务取消功能，允许取消尚未执行的任务：

class CancellableThreadPool {
private:// 可取消的任务struct CancellableTask {std::function<void()> task;std::atomic<bool> cancelled{false};void operator()() {if (!cancelled) {task();}}};std::vector<std::shared_ptr<CancellableTask>> pending_tasks;// ...其他线程池成员...public:// 提交可取消的任务template<class F, class... Args>std::pair<std::future<typename std::result_of<F(Args...)>::type>, std::shared_ptr<CancellableTask>>enqueue_cancellable(F&& f, Args&&... args) {// ...实现可取消任务提交...}// 取消任务bool cancel_task(std::shared_ptr<CancellableTask> task) {if (task) {task->cancelled = true;return true;}return false;}
};

线程池状态监控

添加状态监控功能，以便了解线程池的运行情况：

class MonitorableThreadPool {
private:// ...基本线程池成员...std::atomic<size_t> completed_tasks{0};std::atomic<size_t> active_tasks{0};std::chrono::time_point<std::chrono::steady_clock> start_time;public:// 获取状态信息struct PoolStats {size_t thread_count;size_t active_tasks;size_t completed_tasks;size_t queued_tasks;double uptime_seconds;};PoolStats get_stats() {std::lock_guard<std::mutex> lock(queue_mutex);auto now = std::chrono::steady_clock::now();std::chrono::duration<double> uptime = now - start_time;return {workers.size(),active_tasks.load(),completed_tasks.load(),tasks.size(),uptime.count()};}// 在任务执行前后更新活跃任务计数和完成任务计数// ...
};

性能优化

要构建高性能线程池，需要关注以下几个优化方向：

锁竞争优化

锁竞争是影响线程池性能的主要因素，可以通过以下方式减少锁竞争：

使用无锁数据结构：用无锁队列代替互斥锁保护的标准队列

#include <boost/lockfree/queue.hpp>class LockFreeThreadPool {
private:boost::lockfree::queue<std::function<void()>*> tasks{128};// ...其他成员...
};

分区任务队列：为每个工作线程分配独立的任务队列，减少全局锁争用

class PartitionedThreadPool {
private:struct WorkerData {std::queue<std::function<void()>> local_queue;std::mutex local_mutex;std::condition_variable local_condition;};std::vector<std::unique_ptr<WorkerData>> worker_data;// ...其他成员...
};

批量任务处理：一次从队列中取出多个任务，减少锁操作频率

缓存友好设计

CPU缓存命中率对线程池性能有显著影响：

减少冷缓存启动：工作线程在获取任务后，可以尝试处理多个相关任务，提高缓存局部性

数据对齐：确保关键数据结构按照缓存行对齐，减少缓存行共享

// 确保数据按照64字节缓存行对齐
struct alignas(64) AlignedData {std::atomic<int> counter{0};// ...其他数据...
};

避免虚假共享

虚假共享(False Sharing)是多线程程序的性能杀手：

class FalseSharing {
private:// 糟糕的设计：多个计数器紧密排列，导致虚假共享std::atomic<int> counters[8];// 更好的设计：使用填充避免虚假共享struct PaddedCounter {std::atomic<int> value{0};char padding[60]; // 确保每个计数器占用一个完整的缓存行（64字节）};PaddedCounter better_counters[8];
};

C++17提供了std::hardware_destructive_interference_size来帮助避免虚假共享。

任务批处理

批处理任务可以减少同步开销：

class BatchThreadPool {
private:// ...基本线程池成员...// 批量提交任务template<class Iterator>void enqueue_batch(Iterator begin, Iterator end) {std::lock_guard<std::mutex> lock(queue_mutex);// 一次性将所有任务添加到队列for (auto it = begin; it != end; ++it) {tasks.emplace(*it);}// 唤醒足够多的工作线程size_t count = std::distance(begin, end);for (size_t i = 0; i < std::min(count, workers.size()); ++i) {condition.notify_one();}}
};

线程池应用场景

Web服务器

在Web服务器中，线程池可以高效处理并发HTTP请求：

void web_server_example() {ThreadPool pool(16); // 16个工作线程// 模拟HTTP请求处理auto handle_request = [](int client_id) -> std::string {// 模拟处理HTTP请求std::this_thread::sleep_for(std::chrono::milliseconds(50 + (client_id % 100)));return "Response for client " + std::to_string(client_id);};// 提交100个请求std::vector<std::future<std::string>> responses;for (int i = 0; i < 100; ++i) {responses.push_back(pool.enqueue(handle_request, i));}// 处理响应for (auto& response : responses) {std::cout << response.get() << std::endl;}
}

并行计算

线程池适合执行数据并行的计算任务：

void parallel_computation_example() {const size_t DATA_SIZE = 10000000;std::vector<double> data(DATA_SIZE, 1.0);ThreadPool pool(8);// 计算向量元素的平方根auto process_chunk = [&data](size_t start, size_t end) {for (size_t i = start; i < end; ++i) {data[i] = std::sqrt(data[i]);}};// 将数据分块并提交给线程池std::vector<std::future<void>> futures;const size_t CHUNK_SIZE = DATA_SIZE / 16;for (size_t i = 0; i < DATA_SIZE; i += CHUNK_SIZE) {size_t end = std::min(i + CHUNK_SIZE, DATA_SIZE);futures.push_back(pool.enqueue(process_chunk, i, end));}// 等待所有任务完成for (auto& future : futures) {future.get();}
}

图像处理

线程池可以加速图像处理操作：

struct Pixel { uint8_t r, g, b; };void image_processing_example() {// 模拟图像数据const size_t WIDTH = 1920;const size_t HEIGHT = 1080;std::vector<Pixel> image(WIDTH * HEIGHT);ThreadPool pool(8);// 图像处理函数（此处为简单的灰度转换）auto process_row = [&image, WIDTH](size_t y) {for (size_t x = 0; x < WIDTH; ++x) {Pixel& p = image[y * WIDTH + x];uint8_t gray = static_cast<uint8_t>(0.299 * p.r + 0.587 * p.g + 0.114 * p.b);p.r = p.g = p.b = gray;}};// 按行处理图像std::vector<std::future<void>> futures;for (size_t y = 0; y < HEIGHT; ++y) {futures.push_back(pool.enqueue(process_row, y));}// 等待处理完成for (auto& future : futures) {future.get();}
}

异步IO处理

线程池可以有效管理异步IO操作：

void async_io_example() {ThreadPool pool(4);// 模拟读取文件auto read_file = [](const std::string& filename) -> std::string {// 模拟文件读取延迟std::this_thread::sleep_for(std::chrono::milliseconds(200));return "Content of " + filename;};// 异步读取多个文件std::vector<std::future<std::string>> futures;std::vector<std::string> filenames = {"file1.txt", "file2.txt", "file3.txt", "file4.txt"};for (const auto& filename : filenames) {futures.push_back(pool.enqueue(read_file, filename));}// 处理结果（可以在等待文件读取的同时执行其他操作）for (size_t i = 0; i < futures.size(); ++i) {std::cout << "File " << filenames[i] << ": " << futures[i].get() << std::endl;}
}

实际案例：高性能线程池

完整代码与解析

下面是一个整合了多种优化和高级特性的高性能线程池实现：

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <functional>
#include <vector>
#include <future>
#include <memory>
#include <atomic>
#include <chrono>class AdvancedThreadPool {
public:// 线程池状态enum class State {Running,ShuttingDown,Stopped};// 线程池统计信息struct Stats {size_t thread_count;size_t active_threads;size_t tasks_queued;size_t tasks_completed;double uptime_seconds;};// 构造函数explicit AdvancedThreadPool(size_t thread_count = std::thread::hardware_concurrency()): state_(State::Running),active_threads_(0),tasks_completed_(0),start_time_(std::chrono::steady_clock::now()) {// 创建工作线程workers_.reserve(thread_count);for (size_t i = 0; i < thread_count; ++i) {workers_.emplace_back(&AdvancedThreadPool::worker_thread, this);}}// 析构函数~AdvancedThreadPool() {shutdown();}// 提交任务到线程池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;// 检查线程池状态if (state_ != State::Running) {throw std::runtime_error("Cannot enqueue on non-running thread pool");}// 创建packaged_taskauto task = std::make_shared<std::packaged_task<return_type()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));// 获取futurestd::future<return_type> result = task->get_future();{std::lock_guard<std::mutex> lock(queue_mutex_);tasks_.emplace([task]() { (*task)(); });}condition_.notify_one();return result;}// 平滑关闭线程池void shutdown() {{std::lock_guard<std::mutex> lock(queue_mutex_);if (state_ != State::Running) return;state_ = State::ShuttingDown;}condition_.notify_all();for (std::thread& worker : workers_) {if (worker.joinable()) {worker.join();}}{std::lock_guard<std::mutex> lock(queue_mutex_);state_ = State::Stopped;workers_.clear();// 清空任务队列while (!tasks_.empty()) {tasks_.pop();}}}// 获取统计信息Stats get_stats() const {std::lock_guard<std::mutex> lock(queue_mutex_);auto now = std::chrono::steady_clock::now();std::chrono::duration<double> uptime = now - start_time_;return {workers_.size(),active_threads_.load(),tasks_.size(),tasks_completed_.load(),uptime.count()};}private:// 工作线程函数void worker_thread() {while (true) {std::function<void()> task;{std::unique_lock<std::mutex> lock(queue_mutex_);// 等待任务到来或线程池状态改变condition_.wait(lock, [this] {return state_ != State::Running || !tasks_.empty();});// 检查是否应该退出if (state_ != State::Running && tasks_.empty()) {return;}// 获取任务task = std::move(tasks_.front());tasks_.pop();}// 执行任务active_threads_++;task();tasks_completed_++;active_threads_--;}}// 线程池状态State state_;// 工作线程std::vector<std::thread> workers_;std::queue<std::function<void()>> tasks_;// 同步原语mutable std::mutex queue_mutex_;std::condition_variable condition_;// 统计信息std::atomic<size_t> active_threads_;std::atomic<size_t> tasks_completed_;std::chrono::time_point<std::chrono::steady_clock> start_time_;
};

使用示例

下面是一个综合示例，展示了如何使用上述高性能线程池：

int main() {// 创建线程池AdvancedThreadPool pool(8);// 模拟计算任务auto compute_task = [](int id, int complexity) -> std::string {auto start = std::chrono::high_resolution_clock::now();// 模拟计算double result = 0;for (int i = 0; i < complexity * 1000000; ++i) {result += std::sin(i * 0.0001);}auto end = std::chrono::high_resolution_clock::now();std::chrono::duration<double, std::milli> duration = end - start;return "Task " + std::to_string(id) + " completed in " + std::to_string(duration.count()) + " ms, result: " + std::to_string(result);};// 提交多个不同复杂度的任务std::vector<std::future<std::string>> results;for (int i = 0; i < 20; ++i) {results.push_back(pool.enqueue(compute_task, i, i % 5 + 1));}// 在任务执行过程中查看线程池状态for (int i = 0; i < 5; ++i) {std::this_thread::sleep_for(std::chrono::milliseconds(100));auto stats = pool.get_stats();std::cout << "Pool stats: "<< "threads=" << stats.thread_count<< ", active=" << stats.active_threads<< ", queued=" << stats.tasks_queued<< ", completed=" << stats.tasks_completed<< ", uptime=" << stats.uptime_seconds << "s"<< std::endl;}// 获取并输出所有任务结果for (auto& result : results) {try {std::cout << result.get() << std::endl;} catch (const std::exception& e) {std::cerr << "Task exception: " << e.what() << std::endl;}}// 线程池会在析构时自动关闭return 0;
}

性能测试

线程池的性能测试通常关注以下几个方面：

吞吐量测试：衡量单位时间内能处理的任务数量
延迟测试：衡量任务从提交到完成的时间
可扩展性测试：测试不同线程数下的性能表现
负载测试：在高负载下线程池的表现

void throughput_test(size_t thread_count, size_t task_count) {AdvancedThreadPool pool(thread_count);auto start = std::chrono::high_resolution_clock::now();// 提交大量短任务std::vector<std::future<void>> futures;for (size_t i = 0; i < task_count; ++i) {futures.push_back(pool.enqueue([]() {// 短任务，约1msstd::this_thread::sleep_for(std::chrono::milliseconds(1));}));}// 等待所有任务完成for (auto& f : futures) {f.get();}auto end = std::chrono::high_resolution_clock::now();std::chrono::duration<double> elapsed = end - start;double throughput = task_count / elapsed.count();std::cout << "Thread count: " << thread_count<< ", Tasks: " << task_count<< ", Time: " << elapsed.count() << "s"<< ", Throughput: " << throughput << " tasks/s"<< std::endl;
}int main() {// 测试不同线程数的吞吐量for (size_t threads = 1; threads <= 16; threads *= 2) {throughput_test(threads, 10000);}return 0;
}

注意事项与最佳实践

以下是使用线程池时的一些注意事项和最佳实践：

线程池大小：
- 对于CPU密集型任务，线程数应接近CPU核心数
- 对于IO密集型任务，线程数可以超过CPU核心数
- 避免过多线程，以减少上下文切换开销
任务设计：
- 任务应该是独立的，尽量减少共享状态
- 任务应该有适当的粒度：太小导致调度开销高，太大影响负载均衡
- 避免任务中的阻塞操作，尤其是长时间阻塞
异常处理：
- 任务中的异常应该被捕获并妥善处理
- 使用std::future::get()可以重新抛出任务中的异常
线程安全：
- 线程池自身应该是线程安全的
- 共享资源需要适当的同步机制
- 注意避免死锁，特别是任务中使用多个互斥锁时
性能优化：
- 减少锁竞争，考虑无锁数据结构
- 避免虚假共享
- 利用局部性原理优化缓存使用
生命周期管理：
- 确保线程池的生命周期长于任务的生命周期
- 优雅关闭线程池，等待任务完成
- 避免在析构函数中添加任务