3、线程池的启动2.4、线程池的停止2.5、线程的执行函数run()2.6、任务的运行函数2.7、等待所有线程结束三、测试线程池四、源码地址总结一、前言

decltype。

packaged_task。

make_shared。

mutex。

unique_lock。

notify_one。

future。

queue。

bind。

thread。

2.1、类封装

expireTime):_expireTime(expireTime){}int64_t

_expireTime0;//超时的绝对时间functionvoid()

_func;};typedef

getThreadNum(){unique_lockmutex

lock(_mutex);return

args)-futuredecltype(f(args...)){return

unused.**

它是c里新增的最强大的特性之一它对参数进行了高度泛化它能表示0到任意个数、任意类型的参数

auto

std::futuredecltype(f(args...))

std::futuredecltype(f(args...))返回future调用者可以通过future获取返回值

返回值后置*/templateclass

futuredecltype(f(args...)){//获取现在时间int64_t

expireTime(timeoutMs0)?0:TNOWMStimeoutMs;//

定义返回值类型using

taskmake_sharedpackaged_taskretType()(bind(forwardF(f),forwardArgs(args)...));//

fPtrmake_sharedTaskFunc(expireTime);fPtr-_func[task](){(*task)();};unique_lockmutex

lock(_mutex);//

况再去notify_condition.notify_one();return

task-get_future();}protected:size_t

_mutex;

注意ZERO_ThreadPool::exec执行任务返回的是个future,

int

tpool.exec(std::bind(Test::test,

10);

主要是设置线程池中线程的数量如果线程池已经存在则直接返回防止重复初始化。

bool

lock(_mutex);if(!_threads.empty())return

true;

根据设置的线程数量创建线程并保存在一个数组中。

如果线程池已经存在则直接返回防止重复启动。

bool

lock(_mutex);if(!_threads.empty())return

false;for(size_t

i0;i_threadNum;i){_threads.push_back(new

thread(CPP_ThreadPool::run,this));}return

true;

设置线程退出条件并通知所有线程。

停止时要等待所有线程都执行完任务再销毁线程。

void

注意要有这个{}不然会死锁。

{unique_lockmutex

lock(_mutex);_bTerminatetrue;_condition.notify_all();}size_t

thdCount_threads.size();for(size_t

i0;ithdCount;i){if(_threads[i]-joinable()){_threads[i]-join();}delete

_threads[i];_threads[i]NULL;}unique_lockmutex

读取任务判断任务是否存在如果任务队列为空则进入等待状态直到任务队列不为空或退出线程池这里需要两次判断因为可能存在虚假唤醒。

检测所有任务都是否执行完毕这里使用了原子变量来检测任务是否都执行完原因在于任务队列为空不代表任务已经执行完任务可能还在运行中、也可能是任务刚弹出还没运行使用原子变量来计数就更严谨。

bool

CPP_ThreadPool::get(TaskFuncPtr

task)

lock(_mutex);if(_tasks.empty())//判断任务是否存在{_condition.wait(lock,[this]{return

_bTerminate

!_tasks.empty();//唤醒条件});}if(_bTerminate)return

false;if(!_tasks.empty())//判断任务是否存在{taskmove(_tasks.front());//

使用移动语义_tasks.pop();//弹出一个任务return

true;}return

{while(!isTerminate()){TaskFuncPtr

task;//

okget(task);if(ok){_atomic;try{if(task-_expireTime!0

task-_expireTime

处理超时任务}elsetask-_func();//执行任务}catch(...){}--_atomic;//

任务执行完毕,这里只是为了通知waitForAllDoneunique_lockmutex

_tasks.empty())_condition.notify_all();}}

这里使用了可变模块参数、智能指针、bind、function、捕获列表的相关技术知识。

超时时间

可变模块参数对参数进行了高度泛化它能表示0到任意个数、任意类型的参数。

/**

args)-futuredecltype(f(args...)){return

unused.**

它是c里新增的最强大的特性之一它对参数进行了高度泛化它能表示0到任意个数、任意类型的参数

auto

std::futuredecltype(f(args...))

std::futuredecltype(f(args...))返回future调用者可以通过future获取返回值

返回值后置*/templateclass

futuredecltype(f(args...)){//获取现在时间int64_t

expireTime(timeoutMs0)?0:TNOWMStimeoutMs;//

定义返回值类型using

taskmake_sharedpackaged_taskretType()(bind(forwardF(f),forwardArgs(args)...));//

fPtrmake_sharedTaskFunc(expireTime);fPtr-_func[task](){(*task)();};unique_lockmutex

lock(_mutex);//

况再去notify_condition.notify_one();return

task-get_future();}2.7、等待所有线程结束

bool

CPP_ThreadPool::waitForAllDone(int

millsecond)

lock(_mutex);if(_tasks.empty())return

true;if(millsecond0){_condition.wait(lock,[this]{

return

_condition.wait_for(lock,chrono::milliseconds(millsecond),[this]{

return

threadpool;threadpool.init(2);threadpool.start();//启动线程池//

执行任务threadpool.exec(func1,10);threadpool.exec(func2,20,FLY.);threadpool.exec(1000,func3);threadpool.waitForAllDone();threadpool.stop();

}int

threadpool;threadpool.init(2);threadpool.start();//启动线程池futuredecltype(func1_future(0))

ret01threadpool.exec(func1_future,10);futurestring

ret02threadpool.exec(func2_future,20,FLY.);threadpool.waitForAllDone();coutret01

ret02.get()endl;threadpool.stop();}class

Test{

threadpool;threadpool.init(2);threadpool.start();//启动线程池Test

t1;Test

f1threadpool.exec(bind(Test::test,t1,placeholders::_1),10);auto

f2threadpool.exec(bind(Test::test,t2,placeholders::_1),20);threadpool.waitForAllDone();coutf1

f1.get()endl;coutf2

f2.get()endl;threadpool.stop();}int

main(int