网页制作和网站制作有什么区别,郑州网站优化,wordpress 注册地址,火蝠电商合作需要多少钱什么是池化技术
池化技术是一种资源管理策略#xff0c;它通过重复利用已存在的资源来减少资源的消耗#xff0c;从而提高系统的性能和效率。在计算机编程中#xff0c;池化技术通常用于管理线程、连接、数据库连接等资源。
我们会将可能使用的资源预先创建好#xff0c;…什么是池化技术
池化技术是一种资源管理策略它通过重复利用已存在的资源来减少资源的消耗从而提高系统的性能和效率。在计算机编程中池化技术通常用于管理线程、连接、数据库连接等资源。
我们会将可能使用的资源预先创建好并且将它们创建在一个池中当需要使用这些资源时直接从池中获取使用完毕后再将它们归还到池中而不是每次都创建和销毁资源。
池化技术的引用场景十分广泛,例如线程池、数据库连接池、对象池等今天我们主要要探讨的就是线程池
什么是线程池
线程池是一种典型的池化技术的应用在我们日常使用多线程来处理任务时如果每次都创建和销毁线程,频繁的创建与销毁线程会出现大量不必要的资源消耗降低系统的性能。而在线程池中我们可以预先创建一定数量的线程当需要执行任务时直接从线程池中获取线程来执行任务任务执行完毕后线程并不会被销毁而是继续保留在线程池中等待下一次任务的执行,通过这种线程复用的方式可以大大减少线程的创建和销毁从而提高系统的性能和效率。
线程池的优点 避免频繁创建与销毁线程线程池预先创建并维护一定数量的工作线程避免了频繁创建和销毁线程带来的系统开销特别是在处理大量短生命周期任务时效果尤为显著。 负载均衡与缓存局部性线程池可以根据任务负载动态调整线程工作状态避免过度竞争和闲置。同时线程在执行任务过程中可以充分利用CPU缓存提高执行效率。 控制并发级别通过限制线程池大小和任务队列容量可以有效控制系统的并发级别防止因过度并发导致的资源争抢和性能下降。 简化编程模型 线程池提供了一种简化的编程模型开发者无需关心线程的创建、管理和销毁只需将任务提交给线程池即可。这大大简化了多线程编程的复杂性提高了开发效率。 线程池还提供了一些高级功能如任务优先级、任务超时、任务取消等这些功能可以帮助开发者更好地管理任务执行过程提高系统的可靠性和稳定性。
线程池的组成部分
该线程池类的主要有以下组成部分 线程池类 线程池类主要负责管理线程池的状态并根据线程池的状态淘汰的管理工作线程的状态并且实现任务的异步提交 工作线程类有关线程池状态的枚举类任务队列task_queue用来存放工作线程的列表worker_list控制线程池状态/工作线程列表/以及任务队列的互斥锁用于唤醒/阻塞线程的条件变量基于线程异步实现的任务提交函数 相关定义如下: class ThreadPool{private:class worker_thread; // 工作线程类enum class status_t : std::int8_t{TERMINATED -1,TERMINATING 0,RUNNING 1,PAUSED 2,SHUTDOWN 3}; // 线程池的状态: 已终止:-1,正在终止:0,正在运行:1,已暂停:2,等待线程池中任务完成,但是不接收新任务:3std::atomicstatus_t status; // 线程池的状态std::atomicstd::size_t max_task_count; // 线程池中任务的最大数量std::shared_mutex status_mutex; // 线程池状态互斥锁std::shared_mutex task_queue_mutex; // 任务队列的互斥锁std::shared_mutex worker_lists_mutex; // 工作线程列表的互斥锁std::condition_variable_any task_queue_cv; // 任务队列的条件变量std::condition_variable_any task_queue_cv_full; // 任务队列满的条件变量std::condition_variable_any task_queue_cv_empty; // 任务队列空的条件变量std::queuestd::functionvoid() task_queue; // 任务队列,其中存储待执行的任务std::listworker_thread worker_lists; // 工作线程列表// 考虑到为了确保线程池的唯一性和安全性,禁止使用拷贝赋值与移动赋值ThreadPool(ThreadPool ) delete;ThreadPool operator(ThreadPool ) delete;ThreadPool(ThreadPool ) delete;ThreadPool operator(ThreadPool ) delete;// 在取得对状态变量的访问权后,调用下列函数来改变线程池的状态void pause_with_status_lock(); // 暂停线程池void resume_with_status_lock(); // 恢复线程池void shutdown_with_status_lock(); // 立刻关闭线程池void shutdown_with_wait_status_lock(bool wait_for_tasks); // 等待任务执行完毕关闭线程池void terminate_with_status_lock(); // 终止线程池void wait_with_status_lock(); // 等待所有任务执行完毕public:ThreadPool(std::size_t max_task_count,std::size_t inital_thread_count0); // 构造函数~ThreadPool(); // 析构函数template typename Func, typename... Argsauto submit(Func f, Args ...args) - std::futuredecltype(f(args...)); // 提交任务,实现对线程任务的异步提交void pause(); // 暂停线程池void resume(); // 恢复线程池void shutdown(); // 立刻关闭线程池void shutdown_wait(); // 等待任务执行完毕关闭线程池void terminate(); // 终止线程池void wait(); // 等待所有任务执行完毕void add_thread(std::size_t count); // 增加线程void remove_thread(std::size_t count); // 删除线程void set_max_task_count(std::size_t count);std::size_t get_task_count(); // 获取任务数量std::size_t get_thread_count(); // 获取线程数量};template typename Func, typename... Argsauto ThreadPool::submit(Func f, Args ...args) - std::futuredecltype(f(args...)){std::shared_lockstd::shared_mutex status_lock(status_mutex);switch (status.load()){case status_t::TERMINATED:throw std::runtime_error(ThreadPool is terminated);case status_t::TERMINATING:throw std::runtime_error(ThreadPool is terminating);case status_t::PAUSED:throw std::runtime_error(ThreadPool is paused);case status_t::SHUTDOWN:throw std::runtime_error(ThreadPool is shutdown);case status_t::RUNNING:break;}if(max_task_count 0max_task_count.load()get_task_count())throw std::runtime_error(ThreadPool is full);using return_typedecltype(f(args...));auto taskstd::make_sharedstd::packaged_taskreturn_type()(std::bind(std::forwardFunc(f), std::forwardArgs(args)...));std::unique_lockstd::shared_mutex lock2(task_queue_mutex);task_queue.emplace([task](){ (*task)(); }); lock2.unlock();status_lock.unlock();task_queue_cv.notify_one(); //唤醒一个线程来执行当前任务std::futurereturn_type restask-get_future();return res;}工作线程类 工作线程类主要负责从任务队列中取出任务并执行同时处理根据当前自身线程的状态变化动态调整线程池的状态。 有关工作线程状态的枚举类线程状态的原子变量控制线程阻塞/唤醒的condition_variable 相关定义如下 class ThreadPool::worker_thread{private:enum class status_t:int8_t{TERMINATED-1,TERMINATING0,PAUSE1,RUNNING2,BLOCKED3}; // 1- 线程已终止 0- 线程正在终止 1- 线程已暂停 2- 线程正在运行 3- 线程已阻塞,等待任务中ThreadPool *pool; //指向线程池std::atomicstatus_t status; //线程状态std::shared_mutex status_mutex; //线程状态互斥锁std::binary_semaphore sem; //信号量,要来控制线程的阻塞和唤醒std::thread thread; //工作线程//禁用拷贝构造与移动构造以及相关复赋值worker_thread(const worker_thread ) delete;worker_thread(worker_thread ) delete;worker_thread operator(const worker_thread ) delete;worker_thread operator(worker_thread ) delete;void resume_with_status_lock();status_t terminate_with_status_lock();void pause_with_status_lock();public:worker_thread(ThreadPool *pool);~worker_thread();void pause();void resume();status_t terminate();};线程池的工作机理剖析
对于该项目的线程池而言,它主要存在下面三个机制
基于状态机的状态转换任务的异步执行机制工作线程的任务执行
基于状态机的状态转换
在线程池中主要有两种状态转换,首先是线程池的状态转换,这里我们通过枚举定义了线程池的不同状态,让线程池在这些状态中不断切换,枚举类的定义如下:
enum class status_t : std::int8_t{TERMINATED -1,TERMINATING 0,RUNNING 1,PAUSED 2,SHUTDOWN 3};
// 线程池的状态: 已终止:-1,正在终止:0,正在运行:1,已暂停:2,等待线程池中任务完成,但是不接收新任务:3std::atomicstatus_t status;//线程池的状态,这里使用原子变量来保证操作的原子性实现了线程安全而线程池的状态机模型如下:
具体的状态转化代码可以参考: void ThreadPool::pause_with_status_lock(){switch (status.load()){case status_t::TERMINATED: // 线程池已经终止case status_t::TERMINATING: // 线程池正在终止case status_t::PAUSED: // 线程池已经暂停case status_t::SHUTDOWN: // 线程池已经关闭return;case status_t::RUNNING:status.store(status_t::PAUSED);break;default:throw std::runtime_error(unknown status);}std::unique_lockstd::shared_mutex lock(worker_lists_mutex);for (auto worker : worker_lists){worker.pause();}}工作线程的状态转换就简单了,只有四种状态: enum class status_t:int8_t{TERMINATED-1,TERMINATING0,PAUSE1,RUNNING2,BLOCKED3}; // 1- 线程已终止 0- 线程正在终止 1- 线程已暂停 2- 线程正在运行 3- 线程已阻塞,等待任务中它的状态转换基本上和线程池息息相关,可以参考线程池的状态转换这里不做过多说明。
任务的异步执行
在讲线程的异步执行之前,先来了解一下什么是线程的异步执行:
在计算机科学中“异步”通常指的是一个操作在发起之后不需要等待其完成就可以继续执行后续的操作。异步编程模型允许程序在等待某个耗时操作完成的同时继续执行其他任务从而提高整体效率和响应速度。在c11中为了实现异步执行,引入了std::async,它可以在一个单独的线程中执行一个函数,并返回一个std::future对象,通过这个对象可以获取函数的返回值。这里不做赘述后面我们单独出一篇文章来探讨这个问题这里我们只需要知道它的作用就可以了。这里我们来看一下线程池的任务提交函数submit: template typename Func, typename... Argsauto ThreadPool::submit(Func f, Args ...args) - std::futuredecltype(f(args...)){std::shared_lockstd::shared_mutex status_lock(status_mutex);switch (status.load()){case status_t::TERMINATED:throw std::runtime_error(ThreadPool is terminated);case status_t::TERMINATING:throw std::runtime_error(ThreadPool is terminating);case status_t::PAUSED:throw std::runtime_error(ThreadPool is paused);case status_t::SHUTDOWN:throw std::runtime_error(ThreadPool is shutdown);case status_t::RUNNING:break;}if(max_task_count 0max_task_count.load()get_task_count())throw std::runtime_error(ThreadPool is full);using return_typedecltype(f(args...));auto taskstd::make_sharedstd::packaged_taskreturn_type()(std::bind(std::forwardFunc(f), std::forwardArgs(args)...));std::unique_lockstd::shared_mutex lock2(task_queue_mutex);task_queue.emplace([task](){ (*task)(); }); lock2.unlock();status_lock.unlock();task_queue_cv.notify_one(); //唤醒一个线程来执行当前任务std::futurereturn_type restask-get_future();return res;}它的实现思路如下:
首先查看当前线程池的状态如果不是RUNNING状态,抛出异常查看当前的任务队列是否已满如果已满则抛出异常将任务转换为std::packaged_task对象并将其包装为std::function对象以便在线程池中执行这里使用了std::forward将参数传递给任务函数实现完美转发,通过std::function对象将任务函数包装std::functionvoid()类型,以便在线程池中通过在工作线程中可以用统一的格式直接用 () 进行调用对任何形式的任务进行调用执行。将std::packaged_task对象添加到任务队列中并返回一个std::future对象,该对象可以用于获取任务函数的返回值
工作线程的任务执行
在讲解工作线程的任务执行之前,我们先来看一下它的具体实现
ThreadPool::worker_thread::worker_thread(ThreadPool *pool):pool(pool),status(status_t::RUNNING),sem(0),thread([this](){while (true){// 实现线程状态的判断决定是否由该线程执行任务std::unique_lockstd::shared_mutex unique_lock_status(this-status_mutex);while (true){if (!unique_lock_status.owns_lock()) // 当锁被释放时重新获取锁{unique_lock_status.lock();}bool break_flag false; // 当线程为运行态的时候跳出循环switch (this-status.load()){case status_t::TERMINATING:this-status.store(status_t::TERMINATED);case status_t::TERMINATED:return;case status_t::RUNNING:break_flag true;break;case status_t::PAUSE: // PAUSE状态下需要其他的线程来唤醒该线程需要解锁避免出现死锁unique_lock_status.unlock();this-sem.acquire(); // 阻塞当前线程break;case status_t::BLOCKED: // 不支持Blockeddefault:unique_lock_status.unlock();throw std::runtime_error(invalid status);}if (break_flag){unique_lock_status.unlock();break;}}// 判断队列是否为空如果为空则阻塞当前线程std::unique_lockstd::shared_mutex unique_lock_task(this-pool-task_queue_mutex);while (this-pool-task_queue.empty()){while (true){if (!unique_lock_status.owns_lock()){unique_lock_status.lock();}bool break_flag false;switch (this-status.load()){case status_t::TERMINATING:status.store(status_t::TERMINATED);case status_t::TERMINATED:return;case status_t::PAUSE:unique_lock_task.unlock();unique_lock_status.unlock();this-sem.acquire(); // 阻塞线程break;case status_t::RUNNING:this-status.store(status_t::BLOCKED);case status_t::BLOCKED:break_flag true;break;default:unique_lock_status.unlock();unique_lock_task.unlock();throw std::runtime_error(invalid status);}if (break_flag) // 若为阻塞状态等待唤醒{unique_lock_status.unlock();break;}}this-pool-task_queue_cv.wait(unique_lock_task);while (true){if (!unique_lock_status.owns_lock()){unique_lock_status.lock();}bool break_flag false;switch (this-status.load()){case status_t::TERMINATING:status.store(status_t::TERMINATED);case status_t::TERMINATED:return;case status_t::PAUSE:unique_lock_task.unlock();unique_lock_status.unlock();this-sem.acquire(); // 阻塞线程break;case status_t::BLOCKED:this-status.store(status_t::RUNNING);case status_t::RUNNING:break_flag true;break;default:unique_lock_status.unlock();throw std::runtime_error(invalid status);}if (break_flag) // 若为阻塞状态等待唤醒{unique_lock_status.unlock();break;}}}// 尝试取出任务并执行try{auto task this-pool-task_queue.front();this-pool-task_queue.pop();if (this-pool-task_queue.empty()){this-pool-task_queue_cv_empty.notify_all();}unique_lock_task.unlock();task();}catch (const std::exception e){std::cerre.what()std::endl;}}}){}
工作线程工作的进行是定义在线程构造函数中,即线程开始工作后,会一直执行这个函数,直到线程被销毁,它的实现逻辑是:
确定工作线程状态,决定是否由该线程执行任务查看工作队列是否为空,不为空则取出一个任务执行任务根据线程池状态变更如接收到暂停、恢复、终止等指令工作线程调整自身状态并执行相应操作
拓展:ini文件的读写
考虑到线程池后续想要对其进行拓展,例如加入epoll进行网络通信所以我实现了一个简单的ini解析器他主要有两部分:
Value类:实现了int,double,long,string等基础类型与Value对象之间的互相转换,代码如下
class Value // 封装一个Value对象支持string、int、double、bool类型可以实现多个类型转换为Value对象并且可以转换为多个类型{private:std::string m_value; public://支持各个类型的构造函数Value();Value(const std::string value);Value(const int value);Value(const double value);Value(const bool value);Value(const char* value);~Value();//赋值操作,支持任意类型Value operator(const std::string value);Value operator(const int value);Value operator(const double value);Value operator(const bool value);Value operator(const char* value);//对于值的判断bool operator(const Value other);bool operator!(const Value other);//这里实现可以将Value对象转换为任意类型operator int();operator double();operator bool();operator std::string();};
IniFile:基于嵌套map来存储ini文件的各个分区与分区的键值对实现如下: typedef std::mapstd::string, Value Section;class IniFile{private:std::string m_filename;std::mapstd::string, Section m_sections;std::string trim(std::string s); //去除字符串两边的空格 public:IniFile();IniFile(const std::string filename);~IniFile();bool load(const std::string filename); //加载ini文件bool save(const std::string filename); //保存ini文件void show(); //显示ini文件内容void clear(); //清空ini文件内容Value get(const std::string section,const std::string key); //获取指定分区指定键的值void set(const std::string section,const std::string key,const Value value); //设置指定分区指定键的值bool remove(const std::string section,const std::string key); //删除指定分区指定键的值bool has(const std::string section,const std::string key); //判断键值对是否存在bool has(const std::string section);// 判断分区是否存在Section operator[](const std::string section); // 重载[]操作符,用来访问分区中的键值std::string str(); //将字符串转换为ini文件格式};篇幅有限就不粘详细代码了大家也可以尝试自己实现以下毕竟cpp就是一个不断遭轮子的过程。
结语
线程池的实现是一个复杂的过程需要考虑多线程并发、任务调度、线程管理等多个方面。本文通过一个简单的线程池实现介绍了线程池的基本原理和实现方法。希望本文对您有所帮助。
博客地址教程地址项目地址
