【C++】线程池
C++ 线程池介绍
- 什么是线程池？
线程池（Thread Pool） 是一种并发编程模型，用于管理和复用多个线程，避免频繁创建/销毁线程的开销。它通过预创建一组线程，并将任务提交到队列中，由空闲线程自动执行，从而提升多线程程序的性能和资源利用率。
- 为什么需要线程池？
- 降低开销：线程创建/销毁成本高（涉及系统调用、内存分配）。
- 任务调度优化：避免线程竞争CPU导致的上下文切换损耗。
- 资源管控：限制并发线程数量，防止系统过载。
- 简化编程：将任务提交与执行逻辑解耦。
适用场景：
高并发服务器、批量数据处理、异步任务执行（如日志写入）、GUI后台计算等。
单队列线程池
简单高效，推荐使用！！！
实现
以下是基于C++11/17标准库的线程池简化实现：
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>class ThreadPool {
public:explicit ThreadPool(size_t num_threads = std::thread::hardware_concurrency()) {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(queue_mutex);condition.wait(lock, [this] { return stop || !tasks.empty(); });if (stop && tasks.empty()) return;task = std::move(tasks.front());tasks.pop();}task(); // 执行任务}});}}template<typename F, typename... Args>auto enqueue(F&& f, Args&&... args) -> std::future<decltype(f(args...))> {using return_type = decltype(f(args...));auto task = std::make_shared<std::packaged_task<return_type()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));std::future<return_type> res = task->get_future();{std::unique_lock<std::mutex> lock(queue_mutex);if (stop) throw std::runtime_error("enqueue on stopped ThreadPool");tasks.emplace([task] { (*task)(); });}condition.notify_one(); // 唤醒一个线程return res;}~ThreadPool() {{std::unique_lock<std::mutex> lock(queue_mutex);stop = true;}condition.notify_all();for (std::thread& worker : workers) {if (worker.joinable()) worker.join();}}private:std::vector<std::thread> workers;std::queue<std::function<void()>> tasks;std::mutex queue_mutex;std::condition_variable condition;bool stop = false;
};
线程池核心组件
| 组件 | 作用 |
|---|---|
| 任务队列 | 存储待执行的任务（std::queue<std::function<void()>>） |
| 工作线程组 | 预创建的线程集合，循环从队列中取任务执行（std::vector<std::thread>） |
| 互斥锁 | 保护任务队列的线程安全访问（std::mutex） |
| 条件变量 | 协调线程的休眠与唤醒（std::condition_variable） |
| 停止标志 | 安全终止所有线程（bool stop） |
使用示例
#include <iostream>int main() {ThreadPool pool(4); // 创建4个线程的线程池// 提交多个任务std::vector<std::future<int>> results;for (int i = 0; i < 8; ++i) {results.emplace_back(pool.enqueue([i] {std::this_thread::sleep_for(std::chrono::seconds(1));return i * i;}));}// 获取结果for (auto&& result : results) {std::cout << result.get() << ' ';}std::cout << std::endl;return 0;
}
输出：0 1 4 9 16 25 36 49（执行时间约2秒，4线程并行）
优化方向
- 动态扩缩容：根据任务负载动态调整线程数量。
- 任务优先级：支持优先级队列（如
std::priority_queue）。 - 任务窃取：允许空闲线程从其他线程的任务队列窃取任务。
- 返回值处理：使用
std::future和std::packaged_task传递异步结果。 - 异常处理：捕获任务中的异常并传递到主线程。
注意事项
- 线程安全：确保任务队列的访问通过锁保护。
- 资源释放：析构时需等待所有任务完成，避免资源泄漏。
- 死锁风险：避免在任务中阻塞等待其他任务的结果。
- 性能调优：根据硬件核心数设置合理线程数（通常为
CPU核心数 + 1）。
通过线程池，可以显著提升C++程序的并发性能，同时保持代码的简洁与可维护性。
双队列线程池
总体来说，不推荐！！！
线程池是否可以使用两个队列（空闲队列 & 工作队列）？
在传统线程池设计中，通常使用单个任务队列来管理待执行的任务，所有工作线程共享此队列，并通过条件变量协调任务的获取与执行。但根据实际需求，可以通过两个队列（空闲线程队列 + 工作线程队列）来优化管理，以下是详细分析：
对比
| 设计方式 | 单任务队列 | 双队列（空闲+工作） |
|---|---|---|
| 核心思想 | 所有线程共享一个任务队列，竞争获取任务 | 显式区分空闲线程和工作线程，按需调度 |
| 线程状态管理 | 隐式（通过条件变量等待） | 显式（记录线程是否空闲） |
| 适用场景 | 通用场景，简单高效 | 需要动态扩缩容、优先级调度等复杂场景 |
| 实现复杂度 | 简单 | 较高（需维护线程状态和队列同步） |
双队列设计的意义
- 动态线程管理：
根据任务负载动态调整活跃线程数量，例如：- 当任务队列积压时，从空闲队列唤醒更多线程。
- 当任务减少时，将空闲线程挂起或释放，减少资源占用。
- 优先级调度：
允许为不同优先级的任务分配独立队列，结合线程状态实现更精细的调度。 - 避免过度竞争：
将空闲线程与工作线程分离，减少锁争用（例如：空闲队列单独管理）。
双队列实现示例
以下是一个基于空闲队列和工作队列的线程池框架：
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>class ThreadPool {
public:explicit ThreadPool(size_t max_threads) : max_threads(max_threads) {}void enqueue(std::function<void()> task) {{std::unique_lock<std::mutex> lock(task_mutex);tasks.push(std::move(task));}// 尝试唤醒空闲线程或创建新线程if (!idle_threads.empty()) {std::unique_lock<std::mutex> lock(idle_mutex);auto thread = idle_threads.front();idle_threads.pop();lock.unlock();thread->assign_task(); // 唤醒空闲线程} else if (active_threads < max_threads) {create_thread(); // 创建新线程}}private:size_t max_threads;size_t active_threads = 0;std::queue<std::function<void()>> tasks;std::mutex task_mutex;// 空闲线程队列（存储可复用的线程对象）std::queue<std::shared_ptr<WorkerThread>> idle_threads;std::mutex idle_mutex;class WorkerThread {public:void assign_task(std::function<void()> task) {{std::unique_lock<std::mutex> lock(mtx);current_task = std::move(task);has_task = true;}cv.notify_one();}void run() {while (true) {std::function<void()> task;{std::unique_lock<std::mutex> lock(mtx);cv.wait(lock, [&] { return has_task || terminate; });if (terminate) return;task = std::move(current_task);has_task = false;}task(); // 执行任务// 任务完成后，将自己放回空闲队列pool->return_to_idle(this);}}bool is_idle() const { return !has_task; }private:std::mutex mtx;std::condition_variable cv;std::function<void()> current_task;bool has_task = false;bool terminate = false;};void create_thread() {auto worker = std::make_shared<WorkerThread>();std::thread([worker] { worker->run(); }).detach();active_threads++;}void return_to_idle(WorkerThread* thread) {std::unique_lock<std::mutex> lock(idle_mutex);idle_threads.push(thread);}
};
双队列的优缺点
| 优点 | 缺点 |
|---|---|
| 动态扩缩容：按需增加/减少活跃线程 | 实现复杂度高（需管理线程生命周期） |
| 减少锁竞争（空闲与工作线程分离） | 上下文切换可能增多（线程频繁唤醒/挂起） |
| 支持更复杂的调度策略（如优先级） | 内存占用更高（需维护额外队列） |
适用场景
- 突发性高负载：任务量波动大时，动态调整线程数量。
- 资源敏感环境：需严格控制线程数量（如嵌入式系统）。
- 任务优先级调度：高优先级任务优先分配空闲线程。
总结
- 单队列足够大多数场景：简单高效，推荐优先使用。
- 双队列适合特殊需求：当需要动态扩缩容、优先级调度或减少锁竞争时，可考虑双队列设计，但需权衡实现复杂度与性能收益。
根据具体需求选择设计，避免过度工程化！