24并发设计模式——线程池模式

一、线程池模式介绍

线程池模式（Thread Pool Pattern）是一种并发设计模式，用于管理和循环使用线程资源以处理大量任务。它旨在提高系统性能和资源利用率，特别是在需要频繁创建和销毁线程的环境中。

1、线程池模式结构图

线程池管理器（Thread Pool Manager）：

负责线程池的初始化、维护和管理。它创建线程池中的线程，并根据任务负载调整线程数量。
提供接口供客户端提交任务，并管理任务队列的状态。

线程（Worker Thread）：

线程池中的实际线程，用于执行从任务队列中取出的任务。
线程在任务执行完毕后返回线程池中，等待下一个任务。

任务（Task）：

任务是需要由线程池处理的工作单元。任务通常是实现了特定接口的对象或函数。
线程池的线程从任务队列中取出任务并执行。

任务队列（Task Queue）：

存储待处理的任务。线程池中的线程从任务队列中获取任务并执行。
任务队列的实现可能会影响线程池的性能和效率。

2、创建线程池的4种常见方法

固定大小线程池（Fixed Thread Pool）：

线程池中的线程数量是固定的，适用于任务量和负载相对稳定的场景。

定时任务线程池（Scheduled Thread Pool）：

支持任务的延迟执行和周期性执行。适用于需要在特定时间间隔执行任务的场景。

自适应线程池（Adaptive Thread Pool）：

根据系统负载和任务需求动态调整线程池的大小，能够更好地适应变化的负载。

可缓存线程池（Cached Thread Pool）：

线程池中的线程数量根据任务的需求动态调整。如果线程池中的线程空闲时间超过一定阈值，它们可能会被销毁。

二、线程池的设计方法

示范说明：为了便于演示，所有模拟的任务时间都是固定时间（焊四）。

1、固定大小线程池示例

完整代码

fix_threadpool.cpp

#include <iostream>
#include <vector>
#include <thread>
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <chrono>// 添加互斥锁以保护 std::cout
std::mutex coutMutex;// 线程池
class ThreadPool {
public:ThreadPool(size_t numThreads) : stop(false) {for (size_t i = 0; i < numThreads; ++i) {workers.emplace_back([this] () {while (true) {std::function<void()> task;{std::unique_lock<std::mutex> lock(this->queueMutex);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();}} );}}ThreadPool(const ThreadPool&) = delete;ThreadPool& operator=(const ThreadPool&) = delete;~ThreadPool() {{std::unique_lock<std::mutex> lock(queueMutex);stop = true;}condition.notify_all();for (std::thread& worker : workers) {worker.join(); // 阻塞线程，等待主线程回收子线程}}template<class F, class... Args>void enqueue(F&& f, Args&&... args) {{std::unique_lock<std::mutex> lock(queueMutex);if (stop) {throw std::runtime_error("Error: 在已停止的线程池上排队! ");}tasks.emplace([f, args...]() {f(args...);});}condition.notify_one();}private:std::vector<std::thread> workers;std::queue<std::function<void()>> tasks;std::mutex queueMutex;std::condition_variable condition;bool stop;
};void simulateTasks(int id) {std::lock_guard<std::mutex> guard(coutMutex);std::cout << "线程 " << std::this_thread::get_id() << " 正在执行 " << id << "任务\n";// 模拟任务时间相同，每个任务耗时1msstd::this_thread::sleep_for(std::chrono::milliseconds(1));
}int main() {// 创建4个线程池ThreadPool pool(4);// 创建1081个模拟的任务，加入队列中for (int i = 0; i < 1081; ++i) {pool.enqueue(simulateTasks, i);}return 0;
}

运行效果

代码使用固定线程池创建4个线程，并创建1081个模拟的任务被这4个线程共同并发执行，实际线程号只有4种，他们之间通过互斥锁防止恶性竞争，并通过一把全局互斥锁保护 std::cout 的访问。从运行效果看运行速度非常快。

2、可缓存线程池示例

完整代码

cache_threadpool.cpp

#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <condition_variable>
#include <mutex>
#include <functional>
#include <chrono>
#include <atomic>
#include <future>
#include <map>
#include <iomanip>std::mutex coutMutex; // 添加全局互斥锁以保护 std::coutclass CachedThreadPool {
public:CachedThreadPool(size_t initialThreads = std::thread::hardware_concurrency(), int idleTimeout = 20): maxThreads(initialThreads), shutdown(false), idleTimeout(idleTimeout) {startThreads(initialThreads);}~CachedThreadPool() {{std::unique_lock<std::mutex> lock(queueMutex);shutdown = true;}condVar.notify_all();for (std::thread& worker : workers) {if (worker.joinable()) {worker.join();}}}std::future<void> enqueueTask(std::function<void()> task) {std::packaged_task<void()> packagedTask(std::move(task));std::future<void> future = packagedTask.get_future();{std::unique_lock<std::mutex> lock(queueMutex);tasks.push(std::move(packagedTask));lastActive = std::chrono::steady_clock::now();}condVar.notify_one();return future;}private:void startThreads(size_t count) {for (size_t i = 0; i < count; ++i) {workers.emplace_back([this]() {auto lastActiveTime = std::chrono::steady_clock::now();auto threadStartTime = lastActiveTime; // 记录线程开始活动时间static int TaskCode = 0;while (true) {std::packaged_task<void()> task;{std::unique_lock<std::mutex> lock(queueMutex);// 等待任务到来或者超时(释放互斥锁)condVar.wait_for(lock, std::chrono::seconds(idleTimeout), [this]() {return shutdown || !tasks.empty();});if (!tasks.empty()) {std::lock_guard<std::mutex> guard(coutMutex);task = std::move(tasks.front());tasks.pop();TaskCode++; // 任务代号lastActive = std::chrono::steady_clock::now();std::cout << "线程 " << std::this_thread::get_id() << " 执行任务:" << TaskCode << " 开始时间: " << toTimeString(threadStartTime) << std::endl;std::cout << "线程 " << std::this_thread::get_id()  << " 执行任务:" << TaskCode << " 最后一次活动时间: " << toTimeString(lastActive) << std::endl;lastActiveTime = lastActive;} else {// 当任务为空，从线程池的最后一次活动状态开始记录超时时间auto now = std::chrono::steady_clock::now();while (std::chrono::duration_cast<std::chrono::seconds>(now - lastActiveTime).count() < idleTimeout) {now = std::chrono::steady_clock::now();}if (shutdown && tasks.empty()) {break;}  	}}try {task();} catch (const std::exception& e) {std::cerr << e.what() << std::endl;}}// 线程销毁时的处理{std::lock_guard<std::mutex> guard(coutMutex);std::cout << "线程 " << std::this_thread::get_id() << " 已销毁" << std::endl;}});}}std::string toTimeString(const std::chrono::steady_clock::time_point& tp) const {using namespace std::chrono;auto timeT = system_clock::to_time_t(system_clock::now() + (tp - steady_clock::now()));std::tm tm = *std::localtime(&timeT);std::ostringstream oss;oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S");return oss.str();}size_t maxThreads;std::vector<std::thread> workers;std::queue<std::packaged_task<void()>> tasks;std::mutex queueMutex;std::condition_variable condVar;std::atomic<bool> shutdown;std::chrono::steady_clock::time_point lastActive;int idleTimeout;
};int main() {CachedThreadPool pool;std::vector<std::future<void>> tasks;// 模拟20个任务for (int i = 1; i <= 20; ++i) {tasks.push_back(pool.enqueueTask([i, &pool]() {{std::lock_guard<std::mutex> guard(coutMutex);std::cout << i << " 任务被 " << std::this_thread::get_id() << " 线程执行中." << std::endl;}std::this_thread::sleep_for(std::chrono::seconds(1));}));}// 等待所有任务完成for (auto& task : tasks) {task.get();}// 程序退出前稍作等待，确保所有销毁信息被打印出来std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "\n主线程等待20秒后回收所有线程资源，程序退出 " << std::endl;return 0;
}

运行效果

代码简单实现可缓存的线程池，首先调用std::thread::hardware_concurrency()函数获取4个线程共同完成20个模拟的任务，每个模拟的任务需耗时1s，从输出结果看4个线程4s完成20个模拟任务，完成任务后等待20s触发超时，join()函数阻塞，然后所有子线程被主线程回收，程序退出。

3、定时任务线程池示例

完整代码

timing_threadpool.cpp

#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <functional>
#include <atomic>
#include <iomanip>// 全局一把锁
std::mutex taskMutex;class ThreadPool {
public:static ThreadPool& getInstance(size_t numThreads) {static ThreadPool instance(numThreads);return instance;}ThreadPool(const ThreadPool&) = delete;ThreadPool& operator=(const ThreadPool&) = delete;void enqueueTask(const std::function<void()>& task);void start();void stop();private:ThreadPool(size_t numThreads);~ThreadPool();void workerThread();std::vector<std::thread> workers;std::queue<std::function<void()>> tasks;std::condition_variable condVar;std::atomic<bool> stopFlag;std::atomic<bool> runFlag;std::chrono::seconds interval{10};
};ThreadPool::ThreadPool(size_t numThreads) : stopFlag(false), runFlag(false) {workers.reserve(numThreads);
}ThreadPool::~ThreadPool() { stop(); }void ThreadPool::enqueueTask(const std::function<void()>& task) {std::lock_guard<std::mutex> lock(taskMutex);tasks.push(task);condVar.notify_one();
}void ThreadPool::start() {runFlag = true;for (size_t i = 0; i < workers.capacity(); ++i) {workers.emplace_back(&ThreadPool::workerThread, this);}
}void ThreadPool::stop() {stopFlag = true;condVar.notify_all();for (std::thread& worker : workers) {if (worker.joinable()) {worker.join();}}runFlag = false;
}void ThreadPool::workerThread() {while (!stopFlag) {std::function<void()> task;{std::unique_lock<std::mutex> lock(taskMutex);condVar.wait_for(lock, interval, [this]() {return !tasks.empty() || stopFlag;});if (stopFlag && tasks.empty()) {return;}if (!tasks.empty()) {task = tasks.front();tasks.pop();}}if (task) { task(); }}
}// 任务模拟
void simulateTask(const std::string& type) {std::cout << "线程 " << std::this_thread::get_id() << " 执行任务:" << type <<std::endl;// 模拟任务调度耗时1sstd::this_thread::sleep_for(std::chrono::seconds(1));
}// 时间格式转换
std::string toTimeString(const std::chrono::system_clock::time_point& time) {using namespace std::chrono;auto timeT = system_clock::to_time_t(time);std::tm tm = *std::localtime(&timeT);std::ostringstream oss;oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S");return oss.str();
}int main() {// 单例模式创建2个线程ThreadPool& pool = ThreadPool::getInstance(2);// 启动线程池pool.start();// 添加任务到线程池std::string type = "冒烟测试";std::thread([&pool, type]() {while(true) {pool.enqueueTask([type]() {std::cout << "任务:" << type << ",在 " << toTimeString(std::chrono::system_clock::now()) << " 触发执行" << std::endl;simulateTask(type);});std::this_thread::sleep_for(std::chrono::seconds(10));}}).detach();// 主线程休眠1分钟,不让主线程死循环并留出一分钟时间让子线程周期性执行任务std::this_thread::sleep_for(std::chrono::minutes(1));std::cout << "\n最长演示时间1分钟，线程已回收，程序结束。" << std::endl;pool.stop();return 0;
}

运行效果

定义全局的唯把锁，设计单例模式的线程池，主函数中调用单例创建2个线程加入线程池，这2个线程每隔10s周期性执行simulate函数（函数代表被模拟的任务）。

4、自适应线程池

完整代码

adaptive_threadpool.cpp

#include <iostream>
#include <thread>
#include <functional>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <algorithm>
#include <chrono>
#include <string>
#include <atomic>
#include <iomanip>
#include <sys/select.h>
#include <unistd.h>// 任务锁
std::mutex taskMutex;class ThreadPool {
public:// 获取单例实例static ThreadPool& getInstance(size_t min_threads = 1) {static ThreadPool instance(min_threads);return instance;}ThreadPool(const ThreadPool&) = delete;ThreadPool& operator=(const ThreadPool&) = delete;// 执行任务void execute(std::function<void()> task) {{std::unique_lock<std::mutex> lock(queue_mutex);if (stop) {throw std::runtime_error("线程池已停止运行.");}tasks.emplace(std::move(task));++task_count; // 增加任务计数}condition.notify_one();}void start(size_t min_threads) {std::lock_guard<std::mutex> lock(queue_mutex);max_threads = std::max(max_threads, min_threads);workers.reserve(max_threads);for (size_t i = workers.size(); i < max_threads; ++i) {workers.emplace_back(&ThreadPool::worker_thread, this);}}~ThreadPool() {{std::unique_lock<std::mutex> lock(queue_mutex);stop = true;}condition.notify_all();for (auto& worker : workers) {if (worker.joinable()) {worker.join();}}}// 等待所有任务完成void wait() {std::unique_lock<std::mutex> lock(queue_mutex);finished_condition.wait(lock, [this]() { return task_count == 0; });}private:ThreadPool(size_t min_threads = 1) : max_threads(min_threads), stop(false), task_count(0) {}void worker_thread() {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::lock_guard<std::mutex> lock(queue_mutex);if (tasks.size() > max_threads * 3 && workers.size() < max_threads) {size_t additional_threads = std::min(max_threads - workers.size(), tasks.size() / 10);for (size_t i = 0; i < additional_threads; ++i) {workers.emplace_back(&ThreadPool::worker_thread, this);}}}// 任务完成，减少计数器并通知主线程{std::lock_guard<std::mutex> lock(queue_mutex);--task_count;if (task_count == 0) {finished_condition.notify_one();}}}}std::vector<std::thread> workers;std::queue<std::function<void()>> tasks;std::mutex queue_mutex;std::condition_variable condition;std::condition_variable finished_condition;size_t max_threads;std::atomic<bool> stop;std::atomic<size_t> task_count; // 任务计数器
};// 时间格式转换
std::string toTimeString(const std::chrono::system_clock::time_point& time) {using namespace std::chrono;auto timeT = system_clock::to_time_t(time);std::tm tm = *std::localtime(&timeT);std::ostringstream oss;oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S");return oss.str();
}// 超时处理
bool getInputWithTimeout(std::string& input, int timeoutSeconds) {fd_set set;struct timeval timeout;int rv;FD_ZERO(&set);FD_SET(STDIN_FILENO, &set);timeout.tv_sec = timeoutSeconds;timeout.tv_usec = 0;rv = select(STDIN_FILENO + 1, &set, NULL, NULL, &timeout);if (rv == -1) {std::cerr << "select() 错误" << std::endl;return false;} else if (rv == 0) {return false;} else {std::getline(std::cin, input);return true;}
}// 任务模拟
void simulateTask(const std::string& type, const int index) {std::lock_guard<std::mutex> lock(taskMutex);std::cout << "序号:" << index << " 线程 " << std::this_thread::get_id() << " 在运行任务 " << type << " " << toTimeString(std::chrono::system_clock::now()) << std::endl;
}int main() {
/*模拟客户端永不关机，线程池每隔10s周期性轮询任务,\
!!!  如果任务不存在，或超时，则执行上一次模拟的任务.
*/ThreadPool& pool = ThreadPool::getInstance();// 启动线程池(初始创建2个线程)pool.start(2);static std::string type = "";while (true) {std::cout << "输入任务称号（按 Enter 键结束输入）: " << std::flush;if (getInputWithTimeout(type, 20) && !type.empty()) {std::cout << "\n正在执行新任务 " << type << "\n" << std::endl;} else if (type.empty() && !getInputWithTimeout(type, 20)) {std::cout << "\n输入为空 或 超时20s未输入，执行默认任务.\n" << std::endl;type = "黑神话·悟空";} else {std::cout << "\n输入为空 或 超时20s未输入，执行上一次任务.\n" << std::endl;}// 传递任务参数std::string localType = type;for (int i = 0; i < 20; ++i) {pool.execute([i, localType]() {simulateTask(localType, i);// 模拟任务调度耗时std::this_thread::sleep_for(std::chrono::seconds(i*1));});}// 等待所有任务完成pool.wait();// 所有任务完成后再进行等待std::cout << "\n所有任务已完成，等待 10 秒...\n" << std::endl;std::this_thread::sleep_for(std::chrono::seconds(10));}return 0;
}

代码实现模拟的客户端永不关机，单例线程池每隔10s周期性轮询任务。如果任务不存在，或超时20s终端未输入任何信息，则执行默认的模拟任务，如果任务上次执行过，这次轮询时间不存在新的任务，则执行上一次的任务。

运行效果

三、线程池模式的应用场景

1. Web服务器

请求处理：Web服务器通常需要处理大量并发的HTTP请求。线程池可以管理和复用线程，以高效地处理这些请求。线程池的大小可以根据服务器的负载动态调整，优化资源利用率。

2. 后台任务处理

定时任务：如数据备份、日志清理、定时报告生成等。线程池可以定期执行这些任务，确保它们不会影响系统的响应性能。
异步任务：例如，在用户提交数据后，后台线程可以异步处理数据或执行长时间运行的操作，而不会阻塞用户的操作。

3. 并发数据处理

大数据处理：在大数据处理框架中，如Apache Hadoop、Apache Spark等，线程池用于处理并发的数据任务，如数据读取、转换和写入。
数据分析：在进行数据分析时，线程池可以用来并发处理数据集中的多个数据片段，提高分析效率。

4. 并发计算

计算密集型任务：如图像处理、科学计算等，线程池可以并发执行多个计算任务，缩短计算时间。
模拟和建模：在进行复杂的模拟或建模时，线程池可以同时运行多个模拟实例，加快结果的生成速度。

5. 消息队列处理

消息消费：在消息队列系统（如Kafka、RabbitMQ等）中，线程池可以处理从消息队列中消费的消息。通过线程池的线程并发处理消息，提高系统的吞吐量。

6. 网络应用

连接管理：如数据库连接池和网络连接池，线程池用于管理和复用数据库或网络连接，以提高资源利用率和连接响应速度。
并发请求处理：在需要处理大量并发网络请求的应用中，如在线游戏、实时聊天应用等，线程池可以有效管理网络连接和请求处理。

7. GUI应用

后台任务：在图形用户界面（GUI）应用中，线程池可以处理后台任务（如文件下载、数据加载等），以确保用户界面的响应性和流畅性。
长时间运行的操作：通过线程池处理长时间运行的操作，避免阻塞主线程，从而保持界面的流畅性和响应速度。

8. 实时系统

实时数据处理：在实时系统中，线程池可以并发处理传感器数据、事件流等，以满足实时性要求。
高频交易：在金融交易系统中，线程池可以处理大量并发的交易请求和市场数据，保证交易系统的高效运行。

9. 计算任务调度

批处理作业：在批处理作业调度中，线程池可以调度和并发执行多个作业，优化批处理的执行时间。
任务分配：在分布式系统中，线程池可以管理和分配任务到各个计算节点，协调任务的执行。