class ThreadPool { public: static ThreadPool GetInstance(int size std::thread::hardware_concurrency()) { static ThreadPool pool(size); return pool; } ThreadPool(const ThreadPool) delete; ThreadPool operator(const ThreadPool) delete; // 添加任务到队列 templateclass F, class... Args auto enqueue(F f, Args... args) - std::futuretypename std::result_ofF(Args...)::type { using return_type typename std::result_ofF(Args...)::type; // 创建任务共享指针含packaged_task auto task std::make_sharedstd::packaged_taskreturn_type()(std::bind(std::forwardF(f), std::forwardArgs(args)...)); std::futurereturn_type res task-get_future(); { std::unique_lockstd::mutex lock(queue_mutex); // 禁止在停止后添加任务 if (stop) { throw std::runtime_error(enqueue on stopped ThreadPool); } tasks.emplace([task]() { (*task)(); }); // 封装可调用对象 } condition.notify_one(); // 唤醒一个线程 return res; } void shutdown() { std::unique_lockstd::mutex lock(queue_mutex); stop true; // 设置停止标志 condition.notify_all(); // 唤醒所有线程 // 等待所有线程结束 for (std::thread worker : workers) { worker.join(); } } private: // 构造函数启动size个线程 explicit ThreadPool(int size std::thread::hardware_concurrency()) : stop(false) { for (int i 0; i size; i) { workers.emplace_back([this] { for (;;) { std::functionvoid() task; { std::unique_lockstd::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(); // 执行任务 } }); } } // 获取当前任务队列大小 int getTaskCount() { std::unique_lockstd::mutex lock(queue_mutex); return tasks.size(); } // 获取工作线程数量 int getWorkerCount() { return workers.size(); } // 析构函数等待所有任务完成 ~ThreadPool() { shutdown(); } private: std::vectorstd::thread workers; // 工作线程 std::queuestd::functionvoid() tasks;// 任务队列 std::mutex queue_mutex; // 队列互斥锁 std::condition_variable condition; // 条件变量 std::atomicbool stop; // 停止标志原子操作 }; int test42() { ThreadPool pool(4); // 创建4个线程的线程池 // 提交4个任务并收集future std::vectorstd::futureint results; for (int i 0; i 4; i) { // 每个任务计算 i*10 和 i*101 的和 results.emplace_back(pool.enqueue([](int a, int b, int iSleep) { std::this_thread::sleep_for(std::chrono::seconds( 3 - iSleep)); // 模拟计算耗时 std::cout 任务 a / 10 完成线程ID: std::this_thread::get_id() std::endl; return a b; }, i * 10, i * 10 1, i)); } std::cout 所有任务已提交等待计算结果... std::endl; // 等待所有任务完成并收集结果 for (size_t i 0; i results.size(); i) { std::cout 任务 i 结果: results[i].get() std::endl; } std::cout 所有任务完成 std::endl; return 0; } int test43() { ThreadPool pool ThreadPool::GetInstance(4); // 提交不同类型的任务 std::futureint math_result pool.enqueue([](int a, int b) { std::this_thread::sleep_for(std::chrono::seconds(2)); std::cout 数学计算完成 std::endl; return a * b 100; }, 5, 10); std::futurestd::string string_result pool.enqueue([](const std::string s1, const std::string s2) { std::this_thread::sleep_for(std::chrono::seconds(1)); std::cout 字符串组合完成 std::endl; return s1 s2 !; }, Hello, World); std::futuredouble double_result pool.enqueue([]() { std::this_thread::sleep_for(std::chrono::seconds(3)); std::cout 浮点数计算完成 std::endl; return 3.14159 * 2.0; }); // 等待并获取结果 std::cout 字符串结果: string_result.get() std::endl; std::cout 数学结果: math_result.get() std::endl; std::cout 浮点数结果: double_result.get() std::endl; return 0; }CPU密集型double get_cpu_usage() { // 读取/proc/stat或使用系统API获取CPU使用率 static size_t prev_idle 0, prev_total 0; // ... 实现CPU监控逻辑 return cpu_usage; }IO密集型double calculate_blocking_ratio() { // 阻塞系数 IO等待时间 / (IO等待时间 CPU计算时间) if (avg_task_duration_.count() 0) return 0.8; return static_castdouble(avg_wait_time_.count()) / (avg_wait_time_.count() avg_task_duration_.count()); } size_t calculate_optimal_threads(double blocking_ratio) { size_t cpu_cores std::thread::hardware_concurrency(); // 最优线程数 CPU核心数 / (1 - 阻塞系数) return static_castsize_t(cpu_cores / (1 - blocking_ratio)); } void record_task_metrics(std::chrono::milliseconds duration, std::chrono::milliseconds wait_time) { // 使用滑动窗口计算平均值 avg_task_duration_ update_ema(avg_task_duration_, duration); avg_wait_time_ update_ema(avg_wait_time_, wait_time); }在C线程池设计中核心线程数的确定需要根据任务类型来优化CPU密集型任务核心线程数 CPU核心数 1// 获取CPU核心数 unsigned int cpu_cores std::thread::hardware_concurrency(); int cpu_intensive_threads cpu_cores 1;原因CPU密集型任务几乎一直占用CPU线程过多会导致频繁上下文切换1是为了应对偶尔的线程阻塞如缺页中断IO密集型任务核心线程数 CPU核心数 × 2或更多// 经验公式 int io_intensive_threads cpu_cores * 2; // 或根据阻塞系数精确计算 // 最佳线程数 CPU核心数 / (1 - 阻塞系数) // 阻塞系数通常在0.8~0.9之间原因IO操作时线程会阻塞释放CPU资源更多线程可以充分利用CPU等待IO的时间混合任务场景class ThreadPool { public: ThreadPool(size_t cpu_threads, size_t io_threads) { // 为不同类型任务分配不同线程 for (size_t i 0; i cpu_threads; i) { workers_.emplace_back([this] { cpu_worker(); }); } for (size_t i 0; i io_threads; i) { workers_.emplace_back([this] { io_worker(); }); } } };动态调整策略// 根据任务队列长度动态调整 void adjust_thread_count() { size_t queue_size task_queue_.size(); if (queue_size threshold_high current_threads_ max_threads_) { add_thread(); } else if (queue_size threshold_low current_threads_ min_threads_) { remove_thread(); } }实际建议CPU密集型std::thread::hardware_concurrency()IO密集型2 * std::thread::hardware_concurrency()通用场景先设置为核心数的2倍再根据压测结果调整容器环境注意容器CPU限制不要超过配额最佳实践是通过性能测试来确定最适合你业务场景的线程数。每个类型怎么动态调整核心线程数CPU密集型动态调整class CPUThreadPool { private: size_t min_threads_; size_t max_threads_; size_t active_threads_; std::atomicsize_t idle_threads_{0}; std::chrono::steady_clock::time_point last_adjust_time_; public: CPUThreadPool() : min_threads_(std::thread::hardware_concurrency()), max_threads_(min_threads_ * 2), active_threads_(min_threads_) { // 初始化最小线程数 for (size_t i 0; i min_threads_; i) { create_worker(); } } void dynamic_adjust() { // CPU密集型调整频率要低避免频繁调整 auto now std::chrono::steady_clock::now(); if (now - last_adjust_time_ std::chrono::seconds(5)) { return; } last_adjust_time_ now; // 监控CPU使用率 double cpu_usage get_cpu_usage(); size_t queue_size task_queue_.size(); if (cpu_usage 0.9 queue_size 0) { // CPU使用率过高队列积压适当增加线程 if (active_threads_ max_threads_) { size_t add_count std::min(max_threads_ - active_threads_, size_t(2)); // CPU密集型增加要保守 for (size_t i 0; i add_count; i) { create_worker(); active_threads_; } } } else if (cpu_usage 0.3 idle_threads_ active_threads_ / 2) { // CPU使用率低空闲线程多减少线程 size_t remove_count std::min(idle_threads_.load(), active_threads_ - min_threads_); for (size_t i 0; i remove_count; i) { remove_worker(); active_threads_--; } } } double get_cpu_usage() { // 读取/proc/stat或使用系统API获取CPU使用率 static size_t prev_idle 0, prev_total 0; // ... 实现CPU监控逻辑 return cpu_usage; } };IO密集型动态调整class IOThreadPool { private: size_t min_threads_; size_t max_threads_; size_t active_threads_; std::atomicsize_t pending_io_count_{0}; std::chrono::milliseconds avg_task_duration_{0}; std::chrono::milliseconds avg_wait_time_{0}; public: IOThreadPool() : min_threads_(std::thread::hardware_concurrency()), max_threads_(min_threads_ * 4), // IO密集型可以更大 active_threads_(min_threads_ * 2) { for (size_t i 0; i active_threads_; i) { create_worker(); } } void dynamic_adjust() { // IO密集型可以更频繁调整 auto now std::chrono::steady_clock::now(); if (now - last_adjust_time_ std::chrono::seconds(1)) { return; } last_adjust_time_ now; // 计算阻塞系数 double blocking_ratio calculate_blocking_ratio(); size_t optimal_threads calculate_optimal_threads(blocking_ratio); size_t queue_size task_queue_.size(); size_t queue_growth_rate calculate_queue_growth_rate(); if (queue_growth_rate 10 active_threads_ optimal_threads) { // 队列快速增长快速扩容 size_t add_count std::min(optimal_threads - active_threads_, size_t(5)); // IO密集型可以快速增加 for (size_t i 0; i add_count; i) { create_worker(); active_threads_; } } else if (queue_size 5 active_threads_ optimal_threads) { // 队列空闲渐进缩容 size_t remove_count 1; // IO密集型缩容要慢防止突发流量 for (size_t i 0; i remove_count; i) { if (active_threads_ min_threads_) { remove_worker(); active_threads_--; } } } } double calculate_blocking_ratio() { // 阻塞系数 IO等待时间 / (IO等待时间 CPU计算时间) if (avg_task_duration_.count() 0) return 0.8; return static_castdouble(avg_wait_time_.count()) / (avg_wait_time_.count() avg_task_duration_.count()); } size_t calculate_optimal_threads(double blocking_ratio) { size_t cpu_cores std::thread::hardware_concurrency(); // 最优线程数 CPU核心数 / (1 - 阻塞系数) return static_castsize_t(cpu_cores / (1 - blocking_ratio)); } void record_task_metrics(std::chrono::milliseconds duration, std::chrono::milliseconds wait_time) { // 使用滑动窗口计算平均值 avg_task_duration_ update_ema(avg_task_duration_, duration); avg_wait_time_ update_ema(avg_wait_time_, wait_time); } };混合型自适应线程池class AdaptiveThreadPool { private: struct ThreadMetrics { std::atomicsize_t cpu_tasks_completed{0}; std::atomicsize_t io_tasks_completed{0}; std::chrono::steady_clock::time_point last_active; double cpu_usage; }; std::vectorThreadMetrics thread_metrics_; size_t cpu_cores_; public: AdaptiveThreadPool() : cpu_cores_(std::thread::hardware_concurrency()) { // 初始线程数 size_t initial_threads cpu_cores_ * 2; for (size_t i 0; i initial_threads; i) { thread_metrics_.emplace_back(); create_worker(i); } // 启动监控线程 monitor_thread_ std::thread([this] { monitor_loop(); }); } void monitor_loop() { while (running_) { std::this_thread::sleep_for(std::chrono::seconds(2)); // 分析任务类型分布 auto [cpu_ratio, io_ratio] analyze_task_types(); // 根据任务类型调整策略 if (cpu_ratio 0.7) { // CPU密集型为主保守调整 adjust_for_cpu_intensive(); } else if (io_ratio 0.7) { // IO密集型为主积极调整 adjust_for_io_intensive(); } else { // 混合型平衡调整 adjust_balanced(); } } } std::pairdouble, double analyze_task_types() { size_t total_cpu 0, total_io 0; for (auto metric : thread_metrics_) { total_cpu metric.cpu_tasks_completed; total_io metric.io_tasks_completed; } size_t total total_cpu total_io; if (total 0) return {0.5, 0.5}; return {static_castdouble(total_cpu) / total, static_castdouble(total_io) / total}; } void adjust_for_cpu_intensive() { double cpu_usage get_system_cpu_usage(); size_t optimal cpu_cores_ 1; if (active_threads_ optimal cpu_usage 0.8) { // 缓慢增加 add_threads(1); } else if (active_threads_ optimal cpu_usage 0.5) { // 缓慢减少 remove_threads(1); } } void adjust_for_io_intensive() { double blocking_ratio calculate_blocking_ratio(); size_t optimal static_castsize_t(cpu_cores_ / (1 - blocking_ratio)); optimal std::min(optimal, cpu_cores_ * 8); // 设置上限 size_t queue_size task_queue_.size(); if (active_threads_ optimal queue_size 10) { // 快速扩容 add_threads(std::min(optimal - active_threads_, size_t(5))); } else if (active_threads_ optimal queue_size 2) { // 缓慢缩容 remove_threads(1); } } };关键监控指标struct PoolMetrics { // CPU密集型关注 double cpu_usage; // CPU使用率 size_t active_threads; // 活跃线程数 size_t queue_size; // 队列长度 // IO密集型关注 double blocking_ratio; // 阻塞系数 double avg_response_time; // 平均响应时间 double queue_growth_rate; // 队列增长率 size_t pending_io_count; // 等待IO的任务数 // 通用指标 double thread_utilization; // 线程利用率 size_t rejected_tasks; // 拒绝任务数 }; // 调整策略总结 /* * CPU密集型 * - 调整频率低5-10秒 * - 扩容速度慢每次1-2个线程 * - 缩容速度慢每次1个线程 * - 核心指标CPU使用率、队列长度 * * IO密集型 * - 调整频率高1-2秒 * - 扩容速度快每次3-5个线程 * - 缩容速度慢每次1个线程 * - 核心指标阻塞系数、队列增长率 */create_worker()和remove_worker() 各自代码class ThreadPool { private: std::vectorstd::thread workers_; std::queuestd::functionvoid() task_queue_; std::mutex queue_mutex_; std::condition_variable condition_; std::atomicbool stop_{false}; std::atomicsize_t active_threads_{0}; // 用于标记需要退出的线程 std::queuestd::thread::id threads_to_remove_; std::mutex remove_mutex_; public: // 创建工作线程 void create_worker() { workers_.emplace_back([this] { active_threads_; while (true) { std::functionvoid() task; { std::unique_lockstd::mutex lock(queue_mutex_); // 等待任务或停止信号 condition_.wait(lock, [this] { return stop_ || !task_queue_.empty() || should_exit(); }); // 检查是否需要退出 if (stop_ task_queue_.empty()) { break; } // 检查当前线程是否被标记为移除 if (should_exit() task_queue_.empty()) { remove_self_from_list(); break; } // 获取任务 if (!task_queue_.empty()) { task std::move(task_queue_.front()); task_queue_.pop(); } } // 执行任务 if (task) { task(); } } active_threads_--; }); } // 移除工作线程 void remove_worker() { if (workers_.empty()) return; { std::lock_guardstd::mutex lock(remove_mutex_); // 标记最后一个线程需要退出 threads_to_remove_.push(workers_.back().get_id()); } // 唤醒所有线程让被标记的线程退出 condition_.notify_all(); // 等待线程退出可选异步退出 // 如果需要同步等待可以join该线程 } bool should_exit() { std::lock_guardstd::mutex lock(remove_mutex_); auto current_id std::this_thread::get_id(); if (!threads_to_remove_.empty() threads_to_remove_.front() current_id) { return true; } return false; } void remove_self_from_list() { std::lock_guardstd::mutex lock(remove_mutex_); if (!threads_to_remove_.empty()) { threads_to_remove_.pop(); } } };使用示例int main() { AdvancedThreadPool pool; // 创建初始线程 for (int i 0; i 4; i) { pool.create_worker(); } std::cout Initial threads: pool.get_thread_count() std::endl; // 动态调整 pool.remove_worker(); // 移除1个线程 std::cout After removal: pool.get_thread_count() std::endl; pool.create_worker(); // 添加1个线程 pool.create_worker(); // 再添加1个线程 std::cout After addition: pool.get_thread_count() std::endl; // 批量移除 pool.remove_workers_sync(2); std::cout After batch removal: pool.get_thread_count() std::endl; return 0; }
class ThreadPool { public: static ThreadPool GetInstance(int size std::thread::hardware_concurrency()) { static ThreadPool pool(size); return pool; } ThreadPool(const ThreadPool) delete; ThreadPool operator(const ThreadPool) delete; // 添加任务到队列 templateclass F, class... Args auto enqueue(F f, Args... args) - std::futuretypename std::result_ofF(Args...)::type { using return_type typename std::result_ofF(Args...)::type; // 创建任务共享指针含packaged_task auto task std::make_sharedstd::packaged_taskreturn_type()(std::bind(std::forwardF(f), std::forwardArgs(args)...)); std::futurereturn_type res task-get_future(); { std::unique_lockstd::mutex lock(queue_mutex); // 禁止在停止后添加任务 if (stop) { throw std::runtime_error(enqueue on stopped ThreadPool); } tasks.emplace([task]() { (*task)(); }); // 封装可调用对象 } condition.notify_one(); // 唤醒一个线程 return res; } void shutdown() { std::unique_lockstd::mutex lock(queue_mutex); stop true; // 设置停止标志 condition.notify_all(); // 唤醒所有线程 // 等待所有线程结束 for (std::thread worker : workers) { worker.join(); } } private: // 构造函数启动size个线程 explicit ThreadPool(int size std::thread::hardware_concurrency()) : stop(false) { for (int i 0; i size; i) { workers.emplace_back([this] { for (;;) { std::functionvoid() task; { std::unique_lockstd::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(); // 执行任务 } }); } } // 获取当前任务队列大小 int getTaskCount() { std::unique_lockstd::mutex lock(queue_mutex); return tasks.size(); } // 获取工作线程数量 int getWorkerCount() { return workers.size(); } // 析构函数等待所有任务完成 ~ThreadPool() { shutdown(); } private: std::vectorstd::thread workers; // 工作线程 std::queuestd::functionvoid() tasks;// 任务队列 std::mutex queue_mutex; // 队列互斥锁 std::condition_variable condition; // 条件变量 std::atomicbool stop; // 停止标志原子操作 }; int test42() { ThreadPool pool(4); // 创建4个线程的线程池 // 提交4个任务并收集future std::vectorstd::futureint results; for (int i 0; i 4; i) { // 每个任务计算 i*10 和 i*101 的和 results.emplace_back(pool.enqueue([](int a, int b, int iSleep) { std::this_thread::sleep_for(std::chrono::seconds( 3 - iSleep)); // 模拟计算耗时 std::cout 任务 a / 10 完成线程ID: std::this_thread::get_id() std::endl; return a b; }, i * 10, i * 10 1, i)); } std::cout 所有任务已提交等待计算结果... std::endl; // 等待所有任务完成并收集结果 for (size_t i 0; i results.size(); i) { std::cout 任务 i 结果: results[i].get() std::endl; } std::cout 所有任务完成 std::endl; return 0; } int test43() { ThreadPool pool ThreadPool::GetInstance(4); // 提交不同类型的任务 std::futureint math_result pool.enqueue([](int a, int b) { std::this_thread::sleep_for(std::chrono::seconds(2)); std::cout 数学计算完成 std::endl; return a * b 100; }, 5, 10); std::futurestd::string string_result pool.enqueue([](const std::string s1, const std::string s2) { std::this_thread::sleep_for(std::chrono::seconds(1)); std::cout 字符串组合完成 std::endl; return s1 s2 !; }, Hello, World); std::futuredouble double_result pool.enqueue([]() { std::this_thread::sleep_for(std::chrono::seconds(3)); std::cout 浮点数计算完成 std::endl; return 3.14159 * 2.0; }); // 等待并获取结果 std::cout 字符串结果: string_result.get() std::endl; std::cout 数学结果: math_result.get() std::endl; std::cout 浮点数结果: double_result.get() std::endl; return 0; }CPU密集型double get_cpu_usage() { // 读取/proc/stat或使用系统API获取CPU使用率 static size_t prev_idle 0, prev_total 0; // ... 实现CPU监控逻辑 return cpu_usage; }IO密集型double calculate_blocking_ratio() { // 阻塞系数 IO等待时间 / (IO等待时间 CPU计算时间) if (avg_task_duration_.count() 0) return 0.8; return static_castdouble(avg_wait_time_.count()) / (avg_wait_time_.count() avg_task_duration_.count()); } size_t calculate_optimal_threads(double blocking_ratio) { size_t cpu_cores std::thread::hardware_concurrency(); // 最优线程数 CPU核心数 / (1 - 阻塞系数) return static_castsize_t(cpu_cores / (1 - blocking_ratio)); } void record_task_metrics(std::chrono::milliseconds duration, std::chrono::milliseconds wait_time) { // 使用滑动窗口计算平均值 avg_task_duration_ update_ema(avg_task_duration_, duration); avg_wait_time_ update_ema(avg_wait_time_, wait_time); }在C线程池设计中核心线程数的确定需要根据任务类型来优化CPU密集型任务核心线程数 CPU核心数 1// 获取CPU核心数 unsigned int cpu_cores std::thread::hardware_concurrency(); int cpu_intensive_threads cpu_cores 1;原因CPU密集型任务几乎一直占用CPU线程过多会导致频繁上下文切换1是为了应对偶尔的线程阻塞如缺页中断IO密集型任务核心线程数 CPU核心数 × 2或更多// 经验公式 int io_intensive_threads cpu_cores * 2; // 或根据阻塞系数精确计算 // 最佳线程数 CPU核心数 / (1 - 阻塞系数) // 阻塞系数通常在0.8~0.9之间原因IO操作时线程会阻塞释放CPU资源更多线程可以充分利用CPU等待IO的时间混合任务场景class ThreadPool { public: ThreadPool(size_t cpu_threads, size_t io_threads) { // 为不同类型任务分配不同线程 for (size_t i 0; i cpu_threads; i) { workers_.emplace_back([this] { cpu_worker(); }); } for (size_t i 0; i io_threads; i) { workers_.emplace_back([this] { io_worker(); }); } } };动态调整策略// 根据任务队列长度动态调整 void adjust_thread_count() { size_t queue_size task_queue_.size(); if (queue_size threshold_high current_threads_ max_threads_) { add_thread(); } else if (queue_size threshold_low current_threads_ min_threads_) { remove_thread(); } }实际建议CPU密集型std::thread::hardware_concurrency()IO密集型2 * std::thread::hardware_concurrency()通用场景先设置为核心数的2倍再根据压测结果调整容器环境注意容器CPU限制不要超过配额最佳实践是通过性能测试来确定最适合你业务场景的线程数。每个类型怎么动态调整核心线程数CPU密集型动态调整class CPUThreadPool { private: size_t min_threads_; size_t max_threads_; size_t active_threads_; std::atomicsize_t idle_threads_{0}; std::chrono::steady_clock::time_point last_adjust_time_; public: CPUThreadPool() : min_threads_(std::thread::hardware_concurrency()), max_threads_(min_threads_ * 2), active_threads_(min_threads_) { // 初始化最小线程数 for (size_t i 0; i min_threads_; i) { create_worker(); } } void dynamic_adjust() { // CPU密集型调整频率要低避免频繁调整 auto now std::chrono::steady_clock::now(); if (now - last_adjust_time_ std::chrono::seconds(5)) { return; } last_adjust_time_ now; // 监控CPU使用率 double cpu_usage get_cpu_usage(); size_t queue_size task_queue_.size(); if (cpu_usage 0.9 queue_size 0) { // CPU使用率过高队列积压适当增加线程 if (active_threads_ max_threads_) { size_t add_count std::min(max_threads_ - active_threads_, size_t(2)); // CPU密集型增加要保守 for (size_t i 0; i add_count; i) { create_worker(); active_threads_; } } } else if (cpu_usage 0.3 idle_threads_ active_threads_ / 2) { // CPU使用率低空闲线程多减少线程 size_t remove_count std::min(idle_threads_.load(), active_threads_ - min_threads_); for (size_t i 0; i remove_count; i) { remove_worker(); active_threads_--; } } } double get_cpu_usage() { // 读取/proc/stat或使用系统API获取CPU使用率 static size_t prev_idle 0, prev_total 0; // ... 实现CPU监控逻辑 return cpu_usage; } };IO密集型动态调整class IOThreadPool { private: size_t min_threads_; size_t max_threads_; size_t active_threads_; std::atomicsize_t pending_io_count_{0}; std::chrono::milliseconds avg_task_duration_{0}; std::chrono::milliseconds avg_wait_time_{0}; public: IOThreadPool() : min_threads_(std::thread::hardware_concurrency()), max_threads_(min_threads_ * 4), // IO密集型可以更大 active_threads_(min_threads_ * 2) { for (size_t i 0; i active_threads_; i) { create_worker(); } } void dynamic_adjust() { // IO密集型可以更频繁调整 auto now std::chrono::steady_clock::now(); if (now - last_adjust_time_ std::chrono::seconds(1)) { return; } last_adjust_time_ now; // 计算阻塞系数 double blocking_ratio calculate_blocking_ratio(); size_t optimal_threads calculate_optimal_threads(blocking_ratio); size_t queue_size task_queue_.size(); size_t queue_growth_rate calculate_queue_growth_rate(); if (queue_growth_rate 10 active_threads_ optimal_threads) { // 队列快速增长快速扩容 size_t add_count std::min(optimal_threads - active_threads_, size_t(5)); // IO密集型可以快速增加 for (size_t i 0; i add_count; i) { create_worker(); active_threads_; } } else if (queue_size 5 active_threads_ optimal_threads) { // 队列空闲渐进缩容 size_t remove_count 1; // IO密集型缩容要慢防止突发流量 for (size_t i 0; i remove_count; i) { if (active_threads_ min_threads_) { remove_worker(); active_threads_--; } } } } double calculate_blocking_ratio() { // 阻塞系数 IO等待时间 / (IO等待时间 CPU计算时间) if (avg_task_duration_.count() 0) return 0.8; return static_castdouble(avg_wait_time_.count()) / (avg_wait_time_.count() avg_task_duration_.count()); } size_t calculate_optimal_threads(double blocking_ratio) { size_t cpu_cores std::thread::hardware_concurrency(); // 最优线程数 CPU核心数 / (1 - 阻塞系数) return static_castsize_t(cpu_cores / (1 - blocking_ratio)); } void record_task_metrics(std::chrono::milliseconds duration, std::chrono::milliseconds wait_time) { // 使用滑动窗口计算平均值 avg_task_duration_ update_ema(avg_task_duration_, duration); avg_wait_time_ update_ema(avg_wait_time_, wait_time); } };混合型自适应线程池class AdaptiveThreadPool { private: struct ThreadMetrics { std::atomicsize_t cpu_tasks_completed{0}; std::atomicsize_t io_tasks_completed{0}; std::chrono::steady_clock::time_point last_active; double cpu_usage; }; std::vectorThreadMetrics thread_metrics_; size_t cpu_cores_; public: AdaptiveThreadPool() : cpu_cores_(std::thread::hardware_concurrency()) { // 初始线程数 size_t initial_threads cpu_cores_ * 2; for (size_t i 0; i initial_threads; i) { thread_metrics_.emplace_back(); create_worker(i); } // 启动监控线程 monitor_thread_ std::thread([this] { monitor_loop(); }); } void monitor_loop() { while (running_) { std::this_thread::sleep_for(std::chrono::seconds(2)); // 分析任务类型分布 auto [cpu_ratio, io_ratio] analyze_task_types(); // 根据任务类型调整策略 if (cpu_ratio 0.7) { // CPU密集型为主保守调整 adjust_for_cpu_intensive(); } else if (io_ratio 0.7) { // IO密集型为主积极调整 adjust_for_io_intensive(); } else { // 混合型平衡调整 adjust_balanced(); } } } std::pairdouble, double analyze_task_types() { size_t total_cpu 0, total_io 0; for (auto metric : thread_metrics_) { total_cpu metric.cpu_tasks_completed; total_io metric.io_tasks_completed; } size_t total total_cpu total_io; if (total 0) return {0.5, 0.5}; return {static_castdouble(total_cpu) / total, static_castdouble(total_io) / total}; } void adjust_for_cpu_intensive() { double cpu_usage get_system_cpu_usage(); size_t optimal cpu_cores_ 1; if (active_threads_ optimal cpu_usage 0.8) { // 缓慢增加 add_threads(1); } else if (active_threads_ optimal cpu_usage 0.5) { // 缓慢减少 remove_threads(1); } } void adjust_for_io_intensive() { double blocking_ratio calculate_blocking_ratio(); size_t optimal static_castsize_t(cpu_cores_ / (1 - blocking_ratio)); optimal std::min(optimal, cpu_cores_ * 8); // 设置上限 size_t queue_size task_queue_.size(); if (active_threads_ optimal queue_size 10) { // 快速扩容 add_threads(std::min(optimal - active_threads_, size_t(5))); } else if (active_threads_ optimal queue_size 2) { // 缓慢缩容 remove_threads(1); } } };关键监控指标struct PoolMetrics { // CPU密集型关注 double cpu_usage; // CPU使用率 size_t active_threads; // 活跃线程数 size_t queue_size; // 队列长度 // IO密集型关注 double blocking_ratio; // 阻塞系数 double avg_response_time; // 平均响应时间 double queue_growth_rate; // 队列增长率 size_t pending_io_count; // 等待IO的任务数 // 通用指标 double thread_utilization; // 线程利用率 size_t rejected_tasks; // 拒绝任务数 }; // 调整策略总结 /* * CPU密集型 * - 调整频率低5-10秒 * - 扩容速度慢每次1-2个线程 * - 缩容速度慢每次1个线程 * - 核心指标CPU使用率、队列长度 * * IO密集型 * - 调整频率高1-2秒 * - 扩容速度快每次3-5个线程 * - 缩容速度慢每次1个线程 * - 核心指标阻塞系数、队列增长率 */create_worker()和remove_worker() 各自代码class ThreadPool { private: std::vectorstd::thread workers_; std::queuestd::functionvoid() task_queue_; std::mutex queue_mutex_; std::condition_variable condition_; std::atomicbool stop_{false}; std::atomicsize_t active_threads_{0}; // 用于标记需要退出的线程 std::queuestd::thread::id threads_to_remove_; std::mutex remove_mutex_; public: // 创建工作线程 void create_worker() { workers_.emplace_back([this] { active_threads_; while (true) { std::functionvoid() task; { std::unique_lockstd::mutex lock(queue_mutex_); // 等待任务或停止信号 condition_.wait(lock, [this] { return stop_ || !task_queue_.empty() || should_exit(); }); // 检查是否需要退出 if (stop_ task_queue_.empty()) { break; } // 检查当前线程是否被标记为移除 if (should_exit() task_queue_.empty()) { remove_self_from_list(); break; } // 获取任务 if (!task_queue_.empty()) { task std::move(task_queue_.front()); task_queue_.pop(); } } // 执行任务 if (task) { task(); } } active_threads_--; }); } // 移除工作线程 void remove_worker() { if (workers_.empty()) return; { std::lock_guardstd::mutex lock(remove_mutex_); // 标记最后一个线程需要退出 threads_to_remove_.push(workers_.back().get_id()); } // 唤醒所有线程让被标记的线程退出 condition_.notify_all(); // 等待线程退出可选异步退出 // 如果需要同步等待可以join该线程 } bool should_exit() { std::lock_guardstd::mutex lock(remove_mutex_); auto current_id std::this_thread::get_id(); if (!threads_to_remove_.empty() threads_to_remove_.front() current_id) { return true; } return false; } void remove_self_from_list() { std::lock_guardstd::mutex lock(remove_mutex_); if (!threads_to_remove_.empty()) { threads_to_remove_.pop(); } } };使用示例int main() { AdvancedThreadPool pool; // 创建初始线程 for (int i 0; i 4; i) { pool.create_worker(); } std::cout Initial threads: pool.get_thread_count() std::endl; // 动态调整 pool.remove_worker(); // 移除1个线程 std::cout After removal: pool.get_thread_count() std::endl; pool.create_worker(); // 添加1个线程 pool.create_worker(); // 再添加1个线程 std::cout After addition: pool.get_thread_count() std::endl; // 批量移除 pool.remove_workers_sync(2); std::cout After batch removal: pool.get_thread_count() std::endl; return 0; }
