本文共 5631 字,大约阅读时间需要 18 分钟。
线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小,以默认的优先级运行,并处于多线程单元中。
线程池的组成部分:
代码实现中并没有实现任务接口,而是把Runnable对象加入到线程池管理器(ThreadPool),然后剩下的事情就由线程池管理器(ThreadPool)来完成了。
为了充分利用多核的优势,我们利用多线程来进行任务处理,但线程也同样不能滥用,会带来一下几个问题:
1)线程本身存在开销,系统必须为每个线程分配如栈,TLS(线程局部存储),寄存器等。
2)线程管理会给系统带来开销,context切换同样会给系统带来成本。
3)线程本身是可以重用的资源,不需要每次都进行初始化。
所以往往在使用中,我们无需把线程与task任务进行一对一对应,只需要预先初始化有限的线程个数来处理无限的task任务即可,线程池应运而生,原理也就是如此。
线程池具体构造如下:
主要含有三个队列
工作队列是一个阻塞队列,任务(仿函数)任务不算被push进来(notify阻塞获取的工作线程),工作线程队列(一直不变)则从该队列中获取任务执行(wait获取,当任务队列为空时阻塞等待通知),如果获取到任务,则将线程会进入忙碌线程队列中,执行任务的仿函数,当工作完成,重新移出工作线程队列。
代码实现:
#pragma once#ifndef THREAD_POOL_H#define THREAD_POOL_H#include#include #include #include //#include //#include //#include #include namespace std{//线程池最大容量,应尽量设小一点#define THREADPOOL_MAX_NUM 16//#define THREADPOOL_AUTO_GROW//线程池,可以提交变参函数或拉姆达表达式的匿名函数执行,可以获取执行返回值//不直接支持类成员函数, 支持类静态成员函数或全局函数,Opteron()函数等class threadpool{ using Task = function ; //定义类型 vector _pool; //线程池 queue _tasks; //任务队列 mutex _lock; //同步 condition_variable _task_cv; //条件阻塞 atomic _run{ true }; //线程池是否执行 atomic _idlThrNum{ 0 }; //空闲线程数量public: inline threadpool(unsigned short size = 4) { addThread(size); } inline ~threadpool() { _run=false; _task_cv.notify_all(); // 唤醒所有线程执行 for (thread& thread : _pool) { //thread.detach(); // 让线程“自生自灭” if(thread.joinable()) thread.join(); // 等待任务结束, 前提:线程一定会执行完 } }public: // 提交一个任务 // 调用.get()获取返回值会等待任务执行完,获取返回值 // 有两种方法可以实现调用类成员, // 一种是使用 bind: .commit(std::bind(&Dog::sayHello, &dog)); // 一种是用 mem_fn: .commit(std::mem_fn(&Dog::sayHello), this) template auto commit(F&& f, Args&&... args) ->future { if (!_run) // stoped ?? throw runtime_error("commit on ThreadPool is stopped."); using RetType = decltype(f(args...)); // typename std::result_of ::type, 函数 f 的返回值类型 auto task = make_shared >( bind(forward (f), forward (args)...) ); // 把函数入口及参数,打包(绑定) future future = task->get_future(); { // 添加任务到队列 lock_guard lock{ _lock };//对当前块的语句加锁 lock_guard 是 mutex 的 stack 封装类,构造的时候 lock(),析构的时候 unlock() _tasks.emplace([task](){ // push(Task{...}) 放到队列后面 (*task)(); }); }#ifdef THREADPOOL_AUTO_GROW if (_idlThrNum < 1 && _pool.size() < THREADPOOL_MAX_NUM) addThread(1);#endif // !THREADPOOL_AUTO_GROW _task_cv.notify_one(); // 唤醒一个线程执行 return future; } //空闲线程数量 int idlCount() { return _idlThrNum; } //线程数量 int thrCount() { return _pool.size(); }#ifndef THREADPOOL_AUTO_GROWprivate:#endif // !THREADPOOL_AUTO_GROW //添加指定数量的线程 void addThread(unsigned short size) { for (; _pool.size() < THREADPOOL_MAX_NUM && size > 0; --size) { //增加线程数量,但不超过 预定义数量 THREADPOOL_MAX_NUM _pool.emplace_back( [this]{ //工作线程函数 while (_run) { Task task; // 获取一个待执行的 task { // unique_lock 相比 lock_guard 的好处是:可以随时 unlock() 和 lock() unique_lock lock{ _lock }; _task_cv.wait(lock, [this]{ return !_run || !_tasks.empty(); }); // wait 直到有 task if (!_run && _tasks.empty()) return; task = move(_tasks.front()); // 按先进先出从队列取一个 task _tasks.pop(); } _idlThrNum--; task();//执行任务 _idlThrNum++; } }); _idlThrNum++; } }};}#endif //https://github.com/lzpong/
##C++11语言细节 即使懂原理也不代表能写出程序,上面用了众多c++11的“奇技淫巧”,下面简单描述之。
完
转载地址:http://ugnkb.baihongyu.com/