当前位置: 首页 > article >正文

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

一、线程池模式介绍

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

1、线程池模式结构图

  1. 线程池管理器(Thread Pool Manager)

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

    • 线程池中的实际线程,用于执行从任务队列中取出的任务。
    • 线程在任务执行完毕后返回线程池中,等待下一个任务。
  3. 任务(Task)

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

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

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

  1. 固定大小线程池(Fixed Thread Pool)

    • 线程池中的线程数量是固定的,适用于任务量和负载相对稳定的场景。
  2. 定时任务线程池(Scheduled Thread Pool)

    • 支持任务的延迟执行和周期性执行。适用于需要在特定时间间隔执行任务的场景。
  3. 自适应线程池(Adaptive Thread Pool)

    • 根据系统负载和任务需求动态调整线程池的大小,能够更好地适应变化的负载。
  4. 可缓存线程池(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";
	// 模拟任务时间相同,每个任务耗时1ms
	std::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::cout

class 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;
	// 模拟任务调度耗时1s
	std::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. 计算任务调度

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

四、总结

实际上线程池模式属于GoF一书23种设计模式的一种或多种设计模式组合。线程池的主要优点包括减少创建和销毁线程的开销、提高资源的利用率、简化线程管理。它适用于需要处理大量短时间任务的场景(如高并发)。在实际应用中,线程池模式有助于提高系统的响应速度和吞吐量,减少资源浪费,并使任务处理更为高效和可控。 


http://www.kler.cn/a/289822.html

相关文章:

  • 什么是RAG? LangChain的RAG实践!
  • 【Qt】在 Qt Creator 中使用图片资源方法(含素材网站推荐)
  • 红日靶机(七)笔记
  • PHP API如何使用access_token开放接口有效期
  • C++ QT 工具日志异步分批保存
  • 深度学习服务器租赁AutoDL
  • 台球助教系统小程序源码开发与技术解析
  • LLM大模型学习:LoRA 大模型微调的利器
  • 第三届人工智能与智能信息处理国际学术会议(AIIIP 2024)
  • 在SpringMVC中用fmt标签实现国际化/多语言
  • 装饰器模式(Decorator Pattern)
  • MACOS安装配置前端开发环境
  • 北芯生命持续亏损:产能利用率不理想仍扩产能,销售费用越来越高
  • Python世界:文件自动化备份实践
  • 由一个 SwiftData “诡异”运行时崩溃而引发的钩深索隐(一)
  • 工业交换机如何确保品质
  • glsl着色器学习(四)
  • 日常避坑指南:如何合理利用Swap优化MongoDB内存管理
  • Linux驱动开发基础(IRDA 红外遥控模块)
  • E6000物联网主机:打造智慧楼宇的未来
  • Linux:vim编辑器的基本使用
  • 不小心删除丢失了所有短信?如何在 iPhone 上查找和恢复误删除的短信
  • 6 自研rgbd相机基于rk3566之深度计算库移植及测试
  • Spring Boot集成Spring Cloud Scheduler进行任务调度
  • 如何使用Spoon连接data-integration-server并在服务器上执行转换
  • nginx配置白名单服务