当前位置 博文首页 > 藏经阁:基于C++ 11的线程池简单实现

    藏经阁:基于C++ 11的线程池简单实现

    作者:[db:作者] 时间:2021-07-26 08:52

    ??C++11 加入了线程库,从此告别了标准库不支持并发的历史。然而 c++ 对于多线程的支持还是比较低级,稍微高级一点的用法都需要自己去实现,譬如线程池、信号量等。线程池(thread pool)这个东西,在面试上多次被问到,一般的回答都是:“管理一个任务队列,一个线程队列,然后每次取一个任务分配给一个线程去做,循环往复。” 貌似没有问题吧。但是写起程序来的时候就出问题了。

    ??下面给出线程池的代码
    ??threadpool.h

    #pragma once
    
    #ifndef THREAD_POOL_H
    #define THREAD_POOL_H
    
    #include <vector>
    #include <queue>
    #include <atomic>
    #include <future>
    #include <stdexcept>
    
    
    using namespace std;
    
    class threadpool
    {
    private:
    	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) 
    		{
    			if (thread.joinable())
    			{
    				thread.join(); // 等待任务结束, 前提:线程一定会执行完
    			}
    		}
    	}
    
    public:
    
    	// 提交一个任务,调用.get()获取返回值会等待任务执行完,获取返回值
    	template<class F, class... Args>
    	auto commit(F&& f, Args&&... args) ->future<decltype(f(args...))>
    	{
    		if (!_run)    
    			throw runtime_error("commit on ThreadPool is stopped.");
    
    		using RetType = decltype(f(args...)); 
    		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 };
    			_tasks.emplace([task](){ 
    				(*task)();
    			});
    		}
    
    		_task_cv.notify_one(); // 唤醒一个线程执行
    
    		return future;
    	}
    
    	//空闲线程数量
    	int idlCount() { return _idlThrNum; }
    
    	//线程数量
    	int thrCount() { return _pool.size(); }
    
    private:
    
    	//添加指定数量的线程
    	void addThread(unsigned short size)
    	{
    		//获取硬件可支持的并发线程数量
    		const int thread_max_counts = std::thread::hardware_concurrency();
    
    		for (; _pool.size() < thread_max_counts && size > 0; --size)
    		{   
    			//增加线程数量,但不超过 预定义数量 thread_max_counts
    			_pool.emplace_back( [this]{ //工作线程函数
    				while (_run)
    				{
    					Task task; // 获取一个待执行的 task
    					{
    						unique_lock<mutex> lock{ _lock };
    
    						// wait 直到有 task
    						_task_cv.wait(lock, [this]{
    								return !_run || !_tasks.empty();
    						}); 
    
    						if (!_run && _tasks.empty())
    							return;
    
    						task = move(_tasks.front()); // 按先进先出从队列取一个 task
    						_tasks.pop();
    					}
    
    					_idlThrNum--;
    					task(); //执行任务
    					_idlThrNum++;
    				}
    			});
    
    			_idlThrNum++;
    		}
    	}
    };
    
    #endif  
    

    ??该线程池的实现全部是在头文件做的,不需要cpp文件

    ??main函数测试代码

    #include "threadpool.h"
    #include <iostream>
    #include <windows.h>
    
    using namespace std;
    
    //普通函数
    void fun1(int slp)
    {
    	cout << "  hello, fun1 !  thread id = " << std::this_thread::get_id() << endl;
    
    	if (slp>0) 
    	{
    		cout << " ======= fun1 sleep " << slp << "  =========  thread id = " << std::this_thread::get_id();
    		std::this_thread::sleep_for(std::chrono::milliseconds(slp));
    	}
    }
    
    //仿函数
    struct gfun 
    {
    	int operator()(int n) 
    	{
    		cout << n << "  hello, gfun ! thread id = " << std::this_thread::get_id();
    		return 1;
    	}
    };
    
    //类静态函数
    class A 
    {    
    public:
    	static int Afun(int n = 0) 
    	{
    		cout << n << "  hello, Afun !  " << std::this_thread::get_id() << endl;
    		return n;
    	}
    
    	//函数必须是 static 的才能使用线程池
    	static std::string Bfun(int n, std::string str, char c) 
    	{
    		cout << n << "  hello, Bfun !  "<< str.c_str() <<"  " << (int)c <<"  " << std::this_thread::get_id() << endl;
    		return str;
    	}
    };
    
    int main()
    {
    	try 
    	{
    		threadpool executor{ 50 };
    		A a;
    		std::future<void> ff = executor.commit(fun1, 0);
    		std::future<int> fg = executor.commit(gfun{}, 0);
    		std::future<int> gg = executor.commit(a.Afun, 9999); 
    		std::future<std::string> gh = executor.commit(A::Bfun, 9998, "mult args", 123);
    		std::future<std::string> fh = executor.commit([]()->std::string { cout << "hello, fh !  " << std::this_thread::get_id() << endl; return "hello,fh ret !"; });
    
    		cout << " =======  sleep ========= " << std::this_thread::get_id() << endl;
    		std::this_thread::sleep_for(std::chrono::microseconds(900)
    
    下一篇:没有了