外卖网站设计,住房和城乡建设部网站无在建,济源城乡建设局网站,网页视频如何下载到电脑文章目录 二十一、day211. 线程池实现1.1 完整代码1.2 解释 二十一、day21
我们之前在学习std::future、std::async、std::promise相关的知识时#xff0c;通过std::promise和packaged_task构建了一个可用的线程池#xff0c;可参考文章#xff1a;并发编程#xff08;6通过std::promise和packaged_task构建了一个可用的线程池可参考文章并发编程6——future、promise、async线程池 | 爱吃土豆的个人博客。但只是将代码给出简单做了介绍从本节开始将从头开始学习如何通过构建线程池以及一些其他和线程池有关的知识。
参考
恋恋风辰官方博客
【C】线程池 - AirCL - 博客园
线程池工作原理和实现 - 【C语言版 】C/C_哔哩哔哩_bilibili
基于C11实现的异步线程池【C/C】_哔哩哔哩_bilibili 1. 线程池实现
1.1 完整代码
#ifndef __THREAD_POOL_H__
#define __THREAD_POOL_H__
#include atomic
#include condition_variable
#include future
#include iostream
#include mutex
#include queue
#include thread
#include vectorclass NoneCopy {
public:~NoneCopy() {}
protected:NoneCopy() {}
private:NoneCopy(const NoneCopy) delete;NoneCopy operator(const NoneCopy) delete;
};class ThreadPool : public NoneCopy {
public:// delte拷贝构造和拷贝赋值// ThreadPool(const ThreadPool) delete;// ThreadPool operator(const ThreadPool) delete;// 简单单例模式的实现static ThreadPool instance() {static ThreadPool ins;return ins;}// 对任务(不接受参数并返回 void 的可调用对象)进行包装的包装器用Task作为别名using Task std::packaged_taskvoid();// 析构函数~ThreadPool() {stop();}// 该函数用于插入新任务至队列并返回新任务的futuretemplate class F, class... Argsauto commit(F f, Args... args) -std::futuredecltype(std::forwardF(f)(std::forwardArgs(args)...)) {using RetType decltype(f(args...)); // 使用RetType作为可调用对象返回类型的别名if (stop_.load()) // 线程池是否处于关闭状态return std::futureRetType{};// 将可调用对象和参数用装饰器packaged_task进行包装auto task std::make_sharedstd::packaged_taskRetType()(std::bind(std::forwardF(f), std::forwardArgs(args)...));// 从包装器获取futurestd::futureRetType ret task-get_future();// 在{}内使用lock_guard锁定互斥量并将task使用emplace插入至任务队列{std::lock_guardstd::mutex cv_mt(cv_mt_);tasks_.emplace([task] { (*task)(); });}cv_lock_.notify_one();return ret;}// 返回线程数量int idleThreadCount() {return thread_num_;}
private:ThreadPool(unsigned int num std::thread::hardware_concurrency()): stop_(false) {{if (num 1)thread_num_ 2;elsethread_num_ num;}start();}// 启动线程池void start() {for (int i 0; i thread_num_; i) {// 使用容器存储线程对象pool_.emplace_back([this]() { // 每个线程执行的lamda函数线程池执行的任务相同while (!this-stop_.load()) {Task task;{std::unique_lockstd::mutex cv_mt(cv_mt_);this-cv_lock_.wait(cv_mt, [this] {return this-stop_.load() || !this-tasks_.empty();});if (this-tasks_.empty())return;task std::move(this-tasks_.front());this-tasks_.pop();}this-thread_num_--;task();this-thread_num_;}});}}// 将线程池关闭void stop() {stop_.store(true); // 将判断变量置为truecv_lock_.notify_all(); // 将所有线程唤醒for (auto td : pool_) {if (td.joinable()) { // 打印线程池中所有的线程idstd::cout join thread td.get_id() std::endl;td.join();}}}
private:std::mutex cv_mt_; // 互斥量std::condition_variable cv_lock_; // 条件变量std::atomic_bool stop_; // 布尔类型配合条件变量使用std::atomic_int thread_num_; // 线程数量std::queueTask tasks_; // 任务队列每个任务使用packaged_task进行包装std::vectorstd::thread pool_; // 使用容器存储线程
};
#endif // !__THREAD_POOL_H__1.2 解释
1首先我们需要设计一个基类所有继承此基类的所有类均会 delete 拷贝构造和拷贝赋值如果父类 delete、default、explicit构造函数那么同样会作用于子类基类 NoneCopy 定义如下
class NoneCopy {
public:~NoneCopy(){}
protected:NoneCopy(){}
private:NoneCopy(const NoneCopy) delete;NoneCopy operator(const NoneCopy) delete;
};class ThreadPool : public NoneCopy {}当 ThreadPool 类继承基类 NoneCopy 后会自动继承父类的拷贝控制属性ThreadPool 的拷贝构造和拷贝赋值会隐式 delete。而且我们需要将 ThreadPool 的默认构造函数设置为 private 或 protected确保外部无法直接调用 ThreadPool() 来创建实例只能通过 instance() 方法获取唯一实例。
2因为线程池不能被拷贝也不能被赋值并且必须是单例模型所以接下来我们实现 ThreadPool 的单例我们在前文说过单例模式有三种实现方式1. 通过std::call_once和std::once_flag实现2. 通过静态成员函数实现3. 智能指针双重检测实现。区别在于1和3可以显式定义删除器而2没办法定义删除器但是2的实现方式最简单。我们这里使用第二种方式即可
// 简单单例模式的实现
static ThreadPool instance() {static ThreadPool ins;return ins;
}但注意该方法只能在C11及以后的平台才可以使用因为返回局部静态变量只有在C11及以上是线程安全的 3在线程池中我们需要将任务存储至任务队列中但因为队列要求存储数据元素的类型相同我们这里必须定义任务的类型为
using Task std::packaged_taskvoid();
std::queueTask tasks_;
tasks_.emplace([task] { (*task)(); });队列使用 STL 的 queue 即可。任务队列 tasks_的元素类型是std::packaged_taskvoid()但为什么后面我们在 commit 函数中插入任务时插入的是一个lambda函数 std::packaged_taskvoid() std::packaged_taskvoid() 表示packaged_task接受任何无传递参数并返回 void 的可调用对象。它可以封装任何符合这个签名的可调用对象包括普通函数、函数对象、和 lambda 表达式。 lambda 表达式 [task] { (*task)(); }这个表达式本身是一个可调用对象。它捕获了外部变量 task并在调用时执行 (*task)()这实际上是调用了 packaged_task 封装的函数。 插入到队列 在使用 tasks_.emplace(...) 时实际上在构造一个 std::functionvoid() 类型的对象而这个对象可以被 std::packaged_task 接受。 由于 lambda 表达式符合 void() 的函数签名因此 std::packaged_task 能够接受它并正确地存储。 std::packaged_task 重载了 operator() 运算符并通过重载的 operator() 来执行包装的可调用对象。比如在命令std::packaged_taskvoid(int) task(myFunction); 中执行task()其实就算再执行myFunction()。所以在 lambda 函数中 (*task)() 其实是在调用可执行函数只不过我们将其封装到了一个无返回值且无参数的 lambda 函数中这是为了所有类型的可调用对象都可以通过一个无返回值且无参数的 lambda 函数封装以便存储至任务队列中。 4ThreadPool 的其他私有成员定义如下 std::mutex cv_mt_; // 互斥量std::condition_variable cv_lock_; // 条件变量std::atomic_bool stop_; // 布尔类型配合条件变量使用std::atomic_int thread_num_; // 线程数量std::queueTask tasks_; // 任务队列每个任务使用packaged_task进行包装std::vectorstd::thread pool_; // 使用容器存储线程thread_num_ 是线程池当前可用线程的数量当我们调用一个线程执行任务时thread_num_ 会减一当任务完成后会加一保证线程池不超负载。pool_ 用于存储线程因为线程的创建和销毁存在一定的开销我们需要将线程存放至 vector 中以便复用线程。当线程中的任务执行完毕后线程池会进入 “可联结状态”joinable但 std::thread 对象本身不会自动销毁仍旧保留在 vector 中让线程处于空闲状态等待下一个任务以便复用。stop_ 用于判断线程池是否被销毁默认是 False当 stop_ True 时线程池销毁此时会将任务队列和线程容器全部销毁等所有任务执行完毕后才会销毁。
5ThreadPool 的构造函数被设置为 private这是为了确保外部无法直接调用 ThreadPool() 来创建实例只能通过 instance() 方法获取唯一实例
ThreadPool(unsigned int num std::thread::hardware_concurrency()): stop_(false) {{if (num 1)thread_num_ 2;elsethread_num_ num;}start();
}std::thread::hardware_concurrency() 是 thread 类是一个静态方法用于获取当前计算机的CPU核心数我们可以根据这个结果在线程池中创建出数量相等的线程每个线程独自占有一个CPU核心这些线程就不用分时复用CPU时间片此时程序的并发效率是最高的。 6start 函数的主要功能是启动线程并将线程存储到一个 vector 中进行管理。在线程的回调函数中lambda 函数执行过程中线程的主要任务是从任务队列中取出任务并执行。如果队列中有任务线程会弹出任务并执行如果队列为空线程将挂起等待新的任务被添加被通知notify_one()。这种实现通过条件变量来避免线程在无任务时的忙等问题从而有效减少资源浪费。而部分初学者在实现线程池时可能会采用简单的循环检查方式即当任务队列为空时线程会不断循环检查队列状态。这种方式容易导致线程忙等占用大量 CPU 资源进而造成性能下降和资源浪费。实现代码如下
void start() {for (int i 0; i thread_num_; i) {// 使用容器存储线程对象pool_.emplace_back([this]() { // 每个线程执行的lamda函数线程池执行的任务相同while (!this-stop_.load()) { // 判断线程池是否停止Task task; // 初始化一个接受无参并无返回值可调用对象的packaged_task{std::unique_lockstd::mutex cv_mt(cv_mt_);// 条件变量判断当满足任务队列不为空或者stop_为true并且当前线程被唤醒时退出挂起状态this-cv_lock_.wait(cv_mt, [this] {return this-stop_.load() || !this-tasks_.empty();});// 队列为空那么就只有stop_为true一种情况此时无任务需要处理直接退出if (this-tasks_.empty())return;// 处理队列剩余任务task std::move(this-tasks_.front());this-tasks_.pop();}this-thread_num_--; // 减少一个线程数用来执行task任务task(); // 执行任务this-thread_num_; // task任务在线程内是同步执行的所以当task任务执行完后可用的线程数加一}});}
}线程池会启动数量为thread_num_的线程每个线程执行以下程序
将当前线程插入至pool_容器内进行管理循环判断线程池是否关闭如果不关闭那么该线程将会无止境的进行工作为了避免while循环占用CPU资源使用条件变量挂起当前当前直至满足任务队列不为空或者stop_为true并且当前线程被唤醒时退出挂起状态。挂起状态时当前线程会释放锁让其他线程可以访问共享资源线程被唤醒时当前线程会重新拿取锁 如果队列为空且当前线程被唤醒那就只有一种可能线程池关闭。此时无任务需要处理直接退出如果队列不为空取出任务队列的第一个元素 因为在当前线程内的任务执行是同步的所以在执行任务前需要将可用线程数减一待执行完任务后可用线程数加一。最后循环第二步~第四步
7当线程池被销毁时必须等待线程池中的所有任务执行完毕后才可以销毁资源
~ThreadPool() {stop();
}// 将线程池关闭
void stop() {stop_.store(true); // 将判断变量置为truecv_lock_.notify_all(); // 将所有线程唤醒for (auto td : pool_) {if (td.joinable()) { // 打印线程池中所有的线程idstd::cout join thread td.get_id() std::endl;td.join();}}
}stop函数中我们将停止标记设置为true并且调用条件变量的notify_all唤醒所有线程被唤醒的线程获取任务队列中的任务如果队列为空则该线程直接返回其他线程继续执行任务直至所有任务结束销毁资源。
8 接下来我们需要封装一个接口用于投递任务给线程池我们在上面说过我们需要将任务存储至任务队列中但因为队列要求存储数据元素的类型相同我们这里必须定义任务的类型。我们定义了线程池执行的任务类型为 void(void) 的可调用对象但是现实情况是存在不同类型的任务返回类型已经参数类型军可能不同我们应该如何将其封装为一个 void(void) 对象而不影响其使用功能我们在一开始简单的说了可以通过 packaged_task 重载的 () 运算符进行调用函数但具体过程需要详细说明。
我们可以通过 bind 将一个函数绑定为 void(void) 类型
int functionint(int param) {std::cout param is param std::endl;return 0;
}
void bindfunction() {std::functionint(void) functionv std::bind(functionint, 3);functionv();
}假如我们的可执行任务是 functionint它接受一个类型为 int 的参数返回 int 类型那么我们可以通过 std::bind 以及 std::function 将其绑定为一个 std::functionint(void) 对象我们通过执行 std::function 对象即可执行可执行函数而且 std::function 是类型是 int(void)。因为我们这里手动给 functionint 了一个参数 3所以 functionint 的参数类型是 void但是它的返回类型仍然是 int。而但我们的任务队列要放入返回值为void参数也为void的函数该怎么办呢
我们可以通过 lambda 生成一个返回值和参数都为 void 的函数函数内部调用 functionv 即可我们将上面的函数functionint和调用的参数3打包放入任务队列
void pushtasktoque() {std::functionint(void) functionv std::bind(functionint, 3);using Task std::packaged_taskvoid();std::queueTask taskque;taskque.emplace([functionv]() {functionv();});
}因为我们是通过一个类型为void(void)的 lambda 表达式间接调用的可调用对象我们可以将这个 lambda 投递至任务队列。
当任务执行完成后由 packaged_task 封装的任务会返回 future 对象我们可以通过调用 future 对象的 get() 或 wait() 函数获取任务结果。修改上面的代码使其返回带有任务结果的 future 对象
std::futureint committask() {std::functionint(void) functionv std::bind(functionint, 3);auto taskf std::make_sharedstd::packaged_taskint(void)(functionv);auto res taskf-get_future();using Task std::packaged_taskvoid();std::queueTask taskque;taskque.emplace([taskf]() {(*taskf)();});return res;
}我们将上面这个函数封装为一个适用于多种情况的模板函数为了保证对象类型的不变使用 C 的完美转发
template class F, class... Args
auto commit(F f, Args... args) - std::futuredecltype(std::forwardF(f)(std::forwardArgs(args)...)) {using RetType decltype(f(args...)); // 使用RetType作为可调用对象返回类型的别名if (stop_.load()) // 线程池是否处于关闭状态return std::futureRetType{};// 将可调用对象和参数用装饰器packaged_task进行包装auto task std::make_sharedstd::packaged_taskRetType()(std::bind(std::forwardF(f), std::forwardArgs(args)...));// 从包装器获取futurestd::futureRetType ret task-get_future();// 在{}内使用lock_guard锁定互斥量并将task使用emplace插入至任务队列{std::lock_guardstd::mutex cv_mt(cv_mt_);tasks_.emplace([task] { (*task)(); });}cv_lock_.notify_one();return ret;
}commit 函数执行完后需要返回 std::packaged_task 对象的 future以便获取任务执行结果但是 future 的类型和可调用对象绑定我们不知道任务执行的结果是什么我们应该如何确定函数的返回类型
我们通过 C11 的尾置推导std::futuredecltype(std::forwardF(f)(std::forwardArgs(args)...))decltype会根据根据表达式推断表达式的结果类型我们用future存储这个类型这个future就是返回值类型。
我们在对 thread、async、future源码进行分析的文章中详细介绍了完美转发原理并且单独写了一篇文章对完美转发进行分析可参考文章
并发编程1——线程、thread源码解析 | 爱吃土豆的个人博客
并发编程8—— std::async、std::future 源码解析 | 爱吃土豆的个人博客
C——完美转发引用折叠forward | 爱吃土豆的个人博客 文章转载自： http://www.morning.jwefry.cn.gov.cn.jwefry.cn http://www.morning.sskhm.cn.gov.cn.sskhm.cn http://www.morning.tstwx.cn.gov.cn.tstwx.cn http://www.morning.yxmcx.cn.gov.cn.yxmcx.cn http://www.morning.knnc.cn.gov.cn.knnc.cn http://www.morning.qcfgd.cn.gov.cn.qcfgd.cn http://www.morning.cnkrd.cn.gov.cn.cnkrd.cn http://www.morning.ptmch.com.gov.cn.ptmch.com http://www.morning.qlrwf.cn.gov.cn.qlrwf.cn http://www.morning.trsdm.cn.gov.cn.trsdm.cn http://www.morning.byjwl.cn.gov.cn.byjwl.cn http://www.morning.tqklh.cn.gov.cn.tqklh.cn http://www.morning.tpmnq.cn.gov.cn.tpmnq.cn http://www.morning.trsdm.cn.gov.cn.trsdm.cn http://www.morning.nmfxs.cn.gov.cn.nmfxs.cn http://www.morning.kcnjz.cn.gov.cn.kcnjz.cn http://www.morning.jzykw.cn.gov.cn.jzykw.cn http://www.morning.bpmnh.cn.gov.cn.bpmnh.cn http://www.morning.kmkpm.cn.gov.cn.kmkpm.cn http://www.morning.rklgm.cn.gov.cn.rklgm.cn http://www.morning.qrsrs.cn.gov.cn.qrsrs.cn http://www.morning.cbnjt.cn.gov.cn.cbnjt.cn http://www.morning.nfpkx.cn.gov.cn.nfpkx.cn http://www.morning.xrrjb.cn.gov.cn.xrrjb.cn http://www.morning.cxtbh.cn.gov.cn.cxtbh.cn http://www.morning.dbqcw.com.gov.cn.dbqcw.com http://www.morning.nlqmp.cn.gov.cn.nlqmp.cn http://www.morning.xlyt.cn.gov.cn.xlyt.cn http://www.morning.jmbgl.cn.gov.cn.jmbgl.cn http://www.morning.pfgln.cn.gov.cn.pfgln.cn http://www.morning.jwrcz.cn.gov.cn.jwrcz.cn http://www.morning.wylpy.cn.gov.cn.wylpy.cn http://www.morning.gqdsm.cn.gov.cn.gqdsm.cn http://www.morning.qpsdq.cn.gov.cn.qpsdq.cn http://www.morning.wfttq.cn.gov.cn.wfttq.cn http://www.morning.zzjpy.cn.gov.cn.zzjpy.cn http://www.morning.dmthy.cn.gov.cn.dmthy.cn http://www.morning.djxnn.cn.gov.cn.djxnn.cn http://www.morning.kskpx.cn.gov.cn.kskpx.cn http://www.morning.lkjzz.cn.gov.cn.lkjzz.cn http://www.morning.lctrz.cn.gov.cn.lctrz.cn http://www.morning.bdzps.cn.gov.cn.bdzps.cn http://www.morning.mmtbn.cn.gov.cn.mmtbn.cn http://www.morning.lizpw.com.gov.cn.lizpw.com http://www.morning.sfzwm.cn.gov.cn.sfzwm.cn http://www.morning.tpchy.cn.gov.cn.tpchy.cn http://www.morning.qrnbs.cn.gov.cn.qrnbs.cn http://www.morning.tkrwm.cn.gov.cn.tkrwm.cn http://www.morning.swbhq.cn.gov.cn.swbhq.cn http://www.morning.fhhry.cn.gov.cn.fhhry.cn http://www.morning.zcwzl.cn.gov.cn.zcwzl.cn http://www.morning.bfjyp.cn.gov.cn.bfjyp.cn http://www.morning.yccnj.cn.gov.cn.yccnj.cn http://www.morning.jcyrs.cn.gov.cn.jcyrs.cn http://www.morning.trzmb.cn.gov.cn.trzmb.cn http://www.morning.ygkb.cn.gov.cn.ygkb.cn http://www.morning.xpgwz.cn.gov.cn.xpgwz.cn http://www.morning.kkqgf.cn.gov.cn.kkqgf.cn http://www.morning.tnmmp.cn.gov.cn.tnmmp.cn http://www.morning.bpds.cn.gov.cn.bpds.cn http://www.morning.ffbl.cn.gov.cn.ffbl.cn http://www.morning.zqzzn.cn.gov.cn.zqzzn.cn http://www.morning.dnbhd.cn.gov.cn.dnbhd.cn http://www.morning.gjlxn.cn.gov.cn.gjlxn.cn http://www.morning.tsgxz.cn.gov.cn.tsgxz.cn http://www.morning.sacxbs.cn.gov.cn.sacxbs.cn http://www.morning.mgwdp.cn.gov.cn.mgwdp.cn http://www.morning.bmmhs.cn.gov.cn.bmmhs.cn http://www.morning.tgdys.cn.gov.cn.tgdys.cn http://www.morning.mqtzd.cn.gov.cn.mqtzd.cn http://www.morning.lsjtq.cn.gov.cn.lsjtq.cn http://www.morning.xckrj.cn.gov.cn.xckrj.cn http://www.morning.byzpl.cn.gov.cn.byzpl.cn http://www.morning.qlkjh.cn.gov.cn.qlkjh.cn http://www.morning.sfwcx.cn.gov.cn.sfwcx.cn http://www.morning.yxlhz.cn.gov.cn.yxlhz.cn http://www.morning.xlmgq.cn.gov.cn.xlmgq.cn http://www.morning.plqhb.cn.gov.cn.plqhb.cn http://www.morning.txtgy.cn.gov.cn.txtgy.cn http://www.morning.ysbrz.cn.gov.cn.ysbrz.cn
