首页 > 代码库 > 基于C++11的线程池
基于C++11的线程池
- 1.封装的线程对象
class task : public std::tr1::enable_shared_from_this<task> { public: task():exit_(false){} task( const task & ) = delete; ~task(){} task & operator =( const task &) = delete; void start(); void stop() { exit_ = true; sync_.notify_one(); } void set_job( const std::function<void()> & job, const std::string & file, int line) {//提交任务 { std::unique_lock<std::mutex> lock(mutex_); job_ = job; file_ = file; line_ = line; } sync_.notify_one();//通知主线程有任务要执行.... } void print_job(){ LOG(INFO)<<"sumbit from:"<<file_<<":"<<line_; } private: bool exit_; std::mutex mutex_; std::condition_variable sync_; std::function< void()> job_; //线程执行的任务,线程任意时刻,最多只能执行一个任务。 std::thread::id id_; std::string file_; int line_; };void task::start() { auto job_proxy = [this] (){ id_ = std::this_thread::get_id(); while( !exit_ ) { std::unique_lock<std::mutex> lock(mutex_); if( job_ ) {//有任务了,需要执行任务了 try { job_(); //执行任务的代码 }catch( std::exception & e) { }catch(...) { } job_ = std::function<void()>(); //释放任务绑定的资源,主要为闭包捕获的资源,特别是shared_ptr对象. tasks->job_completed( shared_from_this() ); //任务执行完成,通知线程池 }else{ //没有任务的时候,等待其他线程提交任务。。 sync_.wait(lock); } } }; std::thread t(job_proxy); //创建并启动与task管理的线程 t.detach(); //分离模式,thread对象销毁了,但是其创建的线程还活着。。。 }
- 2.线程池管理对象
class task_pool { public: task_pool(unsigned int pool_size = 128):max_size_(pool_size),stop_all_(true) { } ~task_pool() { } void job_completed( const std::tr1::shared_ptr<task> & t)//回收task对象 { std::lock_guard<std::mutex> lock(mutex_); bool need_to_notify = idle_tasks_.empty() && (!wait_for_running_jobs_.empty()); busying_tasks_.erase(t); idle_tasks_.push_back(t); LOG(INFO)<<"after job_completed, current idle tasks size:"<< idle_tasks_.size() <<" busying tasks size:"<<busying_tasks_.size() <<" wait for running jobs size:"<<wait_for_running_jobs_.size(); if( !busying_tasks_.empty() ){ (*busying_tasks_.begin())->print_job(); } if( need_to_notify )//任务太多了,之前空闲线程使用完了,有任务在等待执行,需要通知 { sync_.notify_one(); } }; //提交任务 void submit_job( const std::function<void()> & job, const std::string file, int line) { if( stop_all_ ) { return; } std::lock_guard<std::mutex> lock(mutex_); bool need_notify = wait_for_running_jobs_.empty(); wait_for_running_jobs_.push(std::make_tuple(job,file,line)); if( need_notify )//等待执行的任务为空时,需要通知,其他情况不需要通知. { sync_.notify_one(); } } void execute_job() { while(true) { std::unique_lock<std::mutex> lock(mutex_); while(!stop_all_ && wait_for_running_jobs_.empty() ) { //等待其他线程提交任务 sync_.wait(lock); } if( stop_all_ ) { return; } while(!stop_all_ && idle_tasks_.empty()) { //有任务要执行,但是没有空闲线程,等待其他任务执行完成。 sync_.wait(lock); } if( stop_all_ ) { return; } //有任务,也有空闲线程了 auto t = get_task(); auto job =wait_for_running_jobs_.front(); wait_for_running_jobs_.pop(); //分发任务到task 线程. t->set_job(std::get<0>(job), std::get<1>(job), std::get<2>(job)); } } void stop_all() { std::lock_guard<std::mutex> lock(mutex_); stop_all_ = true; for( auto t : idle_tasks_ ) { t->stop(); } idle_tasks_.clear(); for( auto t : busying_tasks_ ) { t->stop(); } while(!wait_for_running_jobs_.empty()){ wait_for_running_jobs_.pop(); } sync_.notify_one(); } void start() {// 初始化启动线程池主线程 try { std::thread t( [this]{ execute_job();}); t.detach(); stop_all_ = false; allocate_tasks(); }catch( std::exception & e ) { LOG(FATAL) << "start tasks pool ... error"<<e.what(); } } protected: std::tr1::shared_ptr<task> get_task() { //获取task对象 if( ! idle_tasks_.empty() ) { auto t = *idle_tasks_.begin(); idle_tasks_.pop_front(); //从空闲队列移除 busying_tasks_.insert(t); //加入忙队列 return t; } return std::tr1::shared_ptr<task>(); } void allocate_tasks() //初始化线程池 { for( int i = 0 ; i < max_size_; i ++ ) { std::tr1::shared_ptr<task> t( new task()); try{ t->start(); idle_tasks_.push_back(t); }catch( std::exception & e) { //超过进程最大线程数限制时,会跑出异常。。。 break; } } } private : unsigned int max_size_; std::list < std::tr1::shared_ptr<task> > idle_tasks_; //空闲任务队列 std::set < std::tr1::shared_ptr<task> > busying_tasks_;//正在执行任务的队列 std::queue< std::tuple< std::function<void()> , std::string, int > > wait_for_running_jobs_; //等待执行的任务 std::mutex mutex_; std::condition_variable sync_; bool stop_all_; };
- usage
static task_pool * tasks = nullptr; static std::once_flag init_flag; static std::once_flag finit_flag; void run_job(const std::function<void()> & job , const std::string & file, int line ) { if( tasks != nullptr) tasks->submit_job(job, file,line); } void task_pool_init( unsigned max_task_size) { std::call_once(init_flag,[max_task_size]{ tasks = new task_pool(max_task_size); tasks->start(); }); } void task_pool_finit() { std::call_once(finit_flag,[]{ tasks->stop_all();}); }
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。