线程池的原理及C++线程池的封装实现

﹏ヽ暗。殇╰゛Y 2022-05-08 01:50 362阅读 0赞

# 线程池原理介绍 #

线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。

线程池的组成部分：

*  线程池管理器（ThreadPoolManager）:用于创建并管理线程池
 *  工作线程（WorkThread）: 线程池中线程
 *  任务接口（Task）:每个任务必须实现的接口，以供工作线程调度任务的执行。
 *  任务队列:用于存放没有处理的任务。提供一种缓冲机制。

多线程技术主要解决处理器单元内多个线程执行的问题，它可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力。  
假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。  
  
如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。  
一个线程池包括以下四个基本组成部分：  
1、线程池管理器（ThreadPool）：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；  
2、工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；  
3、任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；  
4、任务队列（taskQueue）：用于存放没有处理的任务。提供一种缓冲机制。  
  
线程池技术正是关注如何缩短或调整T1,T3时间的技术，从而提高服务器程序性能的。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，不会有T1，T3的开销了。  
线程池不仅调整T1,T3产生的时间段，而且它还显著减少了创建线程的数目，看一个例子：  
假设一个服务器一天要处理50000个请求，并且每个请求需要一个单独的线程完成。在线程池中，线程数一般是固定的，所以产生线程总数不会超过线程池中线程的数目，而如果服务器不利用线程池来处理这些请求则线程总数为50000。一般线程池大小是远小于50000。所以利用线程池的服务器程序不会为了创建50000而在处理请求时浪费时间，从而提高效率。

代码实现中并没有实现任务接口，而是把Runnable对象加入到线程池管理器（ThreadPool），然后剩下的事情就由线程池管理器（ThreadPool）来完成了。

为了充分利用多核的优势，我们利用多线程来进行任务处理，但线程也同样不能滥用，会带来一下几个问题：

1）线程本身存在开销，系统必须为每个线程分配如栈，TLS（线程局部存储），寄存器等。

2）线程管理会给系统带来开销，context切换同样会给系统带来成本。

3）线程本身是可以重用的资源，不需要每次都进行初始化。

所以往往在使用中，我们无需把线程与task任务进行一对一对应，只需要预先初始化有限的线程个数来处理无限的task任务即可，线程池应运而生，原理也就是如此。

线程池具体构造如下：

![1364972561_9957.png][]

主要含有三个队列

*  工作队列
 *  工作线程队列
 *  忙碌线程队列

工作队列是一个**阻塞**队列，任务（仿函数）任务不算被push进来（notify阻塞获取的工作线程），工作线程队列（一直不变）则从该队列中获取任务执行（wait获取，当任务队列为空时阻塞等待通知），如果获取到任务，则将线程会进入忙碌线程队列中，执行任务的仿函数，当工作完成，重新移出工作线程队列。

代码实现：

#pragma once
    #ifndef THREAD_POOL_H
    #define THREAD_POOL_H
    
    #include <vector>
    #include <queue>
    #include <atomic>
    #include <future>
    //#include <condition_variable>
    //#include <thread>
    //#include <functional>
    #include <stdexcept>
    
    namespace std
    {
    //线程池最大容量,应尽量设小一点
    #define  THREADPOOL_MAX_NUM 16
    //#define  THREADPOOL_AUTO_GROW
    
    //线程池,可以提交变参函数或拉姆达表达式的匿名函数执行,可以获取执行返回值
    //不直接支持类成员函数, 支持类静态成员函数或全局函数,Opteron()函数等
    class threadpool
    {
    	using Task = function<void()>;	//定义类型
    	vector<thread> _pool;          //线程池
    	queue<Task> _tasks;            //任务队列
    	mutex _lock;                   //同步
    	condition_variable _task_cv;   //条件阻塞
    	atomic<bool> _run{ true };     //线程池是否执行
    	atomic<int>  _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<class F, class... Args>
    	auto commit(F&& f, Args&&... args) ->future<decltype(f(args...))>
    	{
    		if (!_run)    // stoped ??
    			throw runtime_error("commit on ThreadPool is stopped.");
    
    		using RetType = decltype(f(args...)); // typename std::result_of<F(Args...)>::type, 函数 f 的返回值类型
    		auto task = make_shared<packaged_task<RetType()>>(
    			bind(forward<F>(f), forward<Args>(args)...)
    		); // 把函数入口及参数,打包(绑定)
    		future<RetType> future = task->get_future();
    		{    // 添加任务到队列
    			lock_guard<mutex> 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_GROW
    private:
    #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<mutex> 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的“奇技淫巧”，下面简单描述之。

1.  using Task = function<void()> 是类型别名，简化了 typedef 的用法。function<void()> 可以认为是一个函数类型，接受任意原型是 void() 的函数，或是函数对象，或是匿名函数。void() 意思是不带参数，没有返回值。
2.  pool.emplace\_back(\[this\]\{...\}) 和 pool.push\_back(\[this\]\{...\}) 功能一样，只不过前者性能会更好；
3.  pool.emplace\_back(\[this\]\{...\}) 是构造了一个线程对象，执行函数是拉姆达匿名函数 ；
4.  所有对象的初始化方式均采用了 \{\}，而不再使用 () 方式，因为风格不够一致且容易出错；
5.  匿名函数： \[this\]\{...\} 不多说。\[\] 是捕捉器，this 是引用域外的变量 this指针， 内部使用死循环, 由cv\_task.wait(lock,\[this\]\{...\}) 来阻塞线程；
6.  delctype(expr) 用来推断 expr 的类型，和 auto 是类似的，相当于类型占位符，占据一个类型的位置；auto f(A a, B b) -> decltype(a+b) 是一种用法，不能写作 decltype(a+b) f(A a, B b)，为啥？！ c++ 就是这么规定的！
7.  commit 方法是不是略奇葩！可以带任意多的参数，第一个参数是 f，后面依次是函数 f 的参数(*注意:参数要传struct/class的话,建议用pointer,小心变量的作用域*)！ 可变参数模板是 c++11 的一大亮点，够亮！至于为什么是 Arg... 和 arg... ，因为规定就是这么用的！
8.  commit 直接使用智能调用stdcall函数，但有两种方法可以实现调用类成员，一种是使用 bind： .commit(std::bind(&Dog::sayHello, &dog))； 一种是用 mem\_fn： .commit(std::mem\_fn(&Dog::sayHello), &dog)；
9.  make\_shared 用来构造 shared\_ptr 智能指针。用法大体是 shared\_ptr p = make\_shared(4) 然后 \*p == 4 。智能指针的好处就是， 自动 delete ！
10. bind 函数，接受函数 f 和部分参数，返回currying后的匿名函数，譬如 bind(add, 4) 可以实现类似 add4 的函数！
11. forward() 函数，类似于 move() 函数，后者是将参数右值化，前者是... 肿么说呢？大概意思就是：不改变最初传入的类型的引用类型(左值还是左值，右值还是右值)；
12. packaged\_task 就是任务函数的封装类，通过 get\_future 获取 future ， 然后通过 future 可以获取函数的返回值(future.get())；packaged\_task 本身可以像函数一样调用 () ；
13. queue 是队列类， front() 获取头部元素， pop() 移除头部元素；back() 获取尾部元素，push() 尾部添加元素；
14. lock\_guard 是 mutex 的 stack 封装类，构造的时候 lock()，析构的时候 unlock()，是 c++ RAII 的 idea；
15. condition\_variable cv; 条件变量， 需要配合 unique\_lock 使用；unique\_lock 相比 lock\_guard 的好处是：可以随时 unlock() 和 lock()。 cv.wait() 之前需要持有 mutex，wait 本身会 unlock() mutex，如果条件满足则会重新持有 mutex。
16. 最后线程池析构的时候,join() 可以等待任务都执行完在结束,很安全!

完

[1364972561_9957.png]: https://img-my.csdn.net/uploads/201304/03/1364972561_9957.png