C++ 线程池的封装实现

亦凉 2022-08-14 01:58 259阅读 0赞

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

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

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

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

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

线程池具体构造如下：

![1364972561_9957.png][]

主要含有三个队列

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

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

定义线程池专属异常：

**struct** TC\_ThreadPool\_Exception : **public** TC\_Exception

**TC\_ThreadPool\_Exception**( **const** string &buffer) : TC\_Exception(buffer)\{\};

**TC\_ThreadPool\_Exception**( **const** string &buffer, **int** err) : TC\_Exception(buffer, err)\{\};

**~TC\_ThreadPool\_Exception** () **throw** ()\{\};

\};

/\*\*

\* @brief 用通线程池类, 与tc\_functor, tc\_functorwrapper配合使用.

\* 使用方式说明:

\* 1 采用tc\_functorwrapper封装一个调用

\* 2 用tc\_threadpool对调用进行执行

\* 具体示例代码请参见:test/test\_tc\_thread\_pool.cpp

\*/

/\*\*线程池本身继承自锁，可以帮助锁定\*\*/

**class** TC\_ThreadPool : **public** TC\_ThreadLock

**public** :

/\*\*

\* @brief 构造函数

\*/

**TC\_ThreadPool** ();

/\*\*

\* @brief 析构, 会停止所有线程

\*/

**~TC\_ThreadPool** ();

/\*\*

\* @brief 初始化.

\* @param num 工作线程个数

\*/

**void** **init**(size\_t num);

/\*\*

\* @brief 获取线程个数.

\* @return size\_t 线程个数

\*/

size\_t **getThreadNum**() \{  Lock sync(\* **this** ); **return** \_jobthread . size(); \}

/\*\*

\* @brief 获取线程池的任务数( exec添加进去的).

\* @return size\_t 线程池的任务数

\*/

size\_t **getJobNum**() \{  **return** \_jobqueue . size(); \}

/\*\*

\* @brief 停止所有线程

\*/

**void** **stop**();

/\*\*

\* @brief 启动所有线程

\*/

**void** **start**();

/\*\*

\* @brief 启动所有线程并, 执行初始化对象.

\* @param ParentFunctor

\* @param tf

\*/

**template** < **class** **ParentFunctor** >

**void** **start**( **const** TC\_FunctorWrapper < **ParentFunctor** > &tf)

**for** (size\_t i = 0; i < \_jobthread .size(); i++)

\_startqueue . push\_back( **new** TC\_FunctorWrapper < **ParentFunctor** >(tf));

start();

/\*\*

\* @brief 添加对象到线程池执行，该函数马上返回，

\* 线程池的线程执行对象

\*/

**template** < **class** **ParentFunctor** >

**void** **exec**( **const** TC\_FunctorWrapper < **ParentFunctor** > &tf)

\_jobqueue .push\_back( **new** TC\_FunctorWrapper < **ParentFunctor** >(tf));

/\*\*

\* @brief 等待所有工作全部结束(队列无任务, 无空闲线程).

\* @param millsecond 等待的时间( ms), -1:永远等待

\* @return true, 所有工作都处理完毕

\* false,超时退出

\*/

**bool** **waitForAllDone**( **int** millsecond = -1);

**public** :

/\*\*

\* @brief 线程数据基类,所有线程的私有数据继承于该类

\*/

**class** ThreadData

**public** :

/\*\*

\* @brief 构造

\*/

**ThreadData**()\{\};

/\*\*

\* @brief 析够

\*/

**virtual** **~ThreadData**()\{\};

/\*\*

\* @brief 生成数据.

\* @ param T

\* @return ThreadData\*

\*/

**template** < **typename** **T** >

**static** **T** \* **makeThreadData**()

**return** **new** **T** ;

\};

/\*\*

\* @brief 设置线程数据.

\* @param p 线程数据

\*/

**static** **void** **setThreadData**( ThreadData \*p);

/\*\*

\* @brief 获取线程数据.

\* @return ThreadData\* 线程数据

\*/

**static** ThreadData \* **getThreadData**();

/\*\*

\* @brief 设置线程数据, key需要自己维护.

\* @param pkey 线程私有数据key

\* @param p 线程指针

\*/

**static** **void** **setThreadData**(pthread\_key\_t pkey, ThreadData \*p);

/\*\*

\* @brief 获取线程数据, key需要自己维护.

\* @param pkey 线程私有数据key

\* @return 指向线程的ThreadData\*指针

\*/

**static** ThreadData \* **getThreadData**(pthread\_key\_t pkey);

**protected** :

/\*\*

\* @brief 释放资源.

\* @param p

\*/

**static** **void** **destructor**( **void** \*p);

/\*\*

\* @brief 初始化key

\*/

**class** KeyInitialize

**public** :

/\*\*

\* @brief 初始化key

\*/

**KeyInitialize**()

**int** ret = pthread\_key\_create(& TC\_ThreadPool :: *g\_key* , TC\_ThreadPool ::*destructor*);

**if** (ret != 0)

**throw** TC\_ThreadPool\_Exception ( "\[TC\_ThreadPool::KeyInitialize\] pthread\_key\_create error" , ret);

/\*\*

\* @brief 释放key

\*/

**~KeyInitialize**()

pthread\_key\_delete( TC\_ThreadPool :: *g\_key* );

\};

/\*\*

\* @brief 初始化key的控制

\*/

**static** KeyInitialize *g\_key\_initialize* ;

/\*\*

\* @brief 数据key

\*/

**static** pthread\_key\_t *g\_key* ;

**protected** :

/\*\*

\* @brief 线程池中的工作线程

\*/

**class** ThreadWorker : **public** TC\_Thread

**public** :

/\*\*

\* @brief 工作线程构造函数.

\* @ param tpool

\*/

**ThreadWorker**( TC\_ThreadPool \*tpool);

/\*\*

\* @brief 通知工作线程结束

\*/

**void** **terminate**();

**protected** :

/\*\*

\* @brief 运行

\*/

**virtual** **void** **run**();

**protected** :

/\*\*

\* 线程池指针

\*/

TC\_ThreadPool  \* \_tpool ;

/\*\*

\* 是否结束线程

\*/

**bool**  \_bTerminate ;

\};

**protected** :

/\*\*

\* @brief 清除

\*/

**void** **clear**();

/\*\*

\* @brief 获取任务, 如果没有任务, 则为NULL.

\* @return TC\_FunctorWrapperInterface\*

\*/

TC\_FunctorWrapperInterface \* **get**( ThreadWorker \*ptw);

/\*\*

\* @brief 获取启动任务.

\* @return TC\_FunctorWrapperInterface\*

\*/

TC\_FunctorWrapperInterface \* **get**();

/\*\*

\* @brief 空闲了一个线程.

\* @param ptw

\*/

**void** **idle**( ThreadWorker \*ptw);

/\*\*

\* @brief 通知等待在任务队列上的工作线程醒来

\*/

**void** **notifyT**();

/\*\*

\* @brief 是否处理结束.

\* @return bool

\*/

**bool** **finish**();

/\*\*

\* @brief 线程退出时调用

\*/

**void** **exit**();

**friend** **class** ThreadWorker ;

**protected** :

/\*\*

\* 任务队列

\*/

TC\_ThreadQueue < TC\_FunctorWrapperInterface \*> \_jobqueue ;

/\*\*

\* 启动任务

\*/

TC\_ThreadQueue < TC\_FunctorWrapperInterface \*> \_startqueue ;

/\*\*

\* 工作线程

\*/

std::vector< ThreadWorker \*>  \_jobthread ;

/\*\*

\* 繁忙线程

\*/

std::set< ThreadWorker \*>  \_busthread ;

/\*\*

\* 任务队列的锁

\*/

TC\_ThreadLock  \_tmutex ;

/\*\*

\* 是否所有任务都执行完毕

\*/

**bool**  \_bAllDone ;

\};

工作线程设计如下：

**TC\_ThreadPool****::ThreadWorker::ThreadWorker**( TC\_ThreadPool \*tpool)

:  \_tpool (tpool)

,  \_bTerminate ( **false** )

**void** **TC\_ThreadPool****::ThreadWorker::terminate**()

\_bTerminate  =  **true** ;

\_tpool \->notifyT();

**void** **TC\_ThreadPool****::ThreadWorker::run**()

//调用初始化部分

TC\_FunctorWrapperInterface  \*pst =  \_tpool \->get();

**if** (pst)

**try**

(\*pst)();

**catch**  ( ... )

**delete**  pst;

pst = NULL;

//调用处理部分

**while**  (! \_bTerminate )

TC\_FunctorWrapperInterface  \*pfw =  \_tpool \->get( **this** );

**if** (pfw != NULL)

auto\_ptr< TC\_FunctorWrapperInterface > apfw(pfw);

**try**

(\*pfw)();

**catch**  ( ... )

\_tpool \->idle( **this** );

//结束

\_tpool \->exit();

每个工作线程在刚开始时都会执行一下初始化操作，并进入一个无限循环的部分 //调用处理部分

**while**  (! \_bTerminate )

TC\_FunctorWrapperInterface  \*pfw =  \_tpool \->get( **this** );

**if** (pfw != NULL)

auto\_ptr< TC\_FunctorWrapperInterface > apfw(pfw);

**try**

(\*pfw)();

**catch**  ( ... )

\_tpool \->idle( **this** );

该工作主要是无限的从线程池的工作队列中获取任务并执行，如果成功获取任务，则会将线程移进忙碌队列：

TC\_FunctorWrapperInterface \***TC\_ThreadPool::****get**( ThreadWorker \*ptw)

TC\_FunctorWrapperInterface  \*pFunctorWrapper = NULL;

**if** (! \_jobqueue .pop\_front(pFunctorWrapper, 1000))

**return** NULL;

Lock  sync( \_tmutex );

\_busthread . insert(ptw);

**return**  pFunctorWrapper;

执行完，移回工作线程队列： \_tpool \->idle( **this** );

**void** **TC\_ThreadPool::****idle**( ThreadWorker \*ptw)

Lock  sync( \_tmutex );

\_busthread .erase(ptw);

//无繁忙线程, 通知等待在线程池结束的线程醒过来

**if** ( \_busthread .empty())

\_bAllDone  =  **true** ;

\_tmutex .notifyAll();

此处jobThread队列初始化后不会改变（因为没有实现自增长功能），所以非线程安全的vector队列即可，busthread的忙碌线程队列会被移进移出，但是操作会自带 Lock  sync( \_tmutex )，该互斥量是线程池本身继承的，所以是共有的，也无需另外使用线程安全的 TC\_ThreadQueue， 使用vector即可。

**TC\_ThreadPool::** **idle** 中的

**if** ( \_busthread .empty())

\_bAllDone  =  **true** ;

\_tmutex .notifyAll();

主要用于当线程池工作起来后的 waitForAllDone 方法:

**bool** **TC\_ThreadPool::****waitForAllDone**( **int**  millsecond)

Lock  sync( \_tmutex );

start1:

//任务队列和繁忙线程都是空的

**if**  (finish())

**return**  **true** ;

//永远等待

**if** (millsecond < 0)

\_tmutex .timedWait(1000);

**goto**  start1;

int64\_t  iNow =  TC\_Common ::*now2ms*();

**int**  m = millsecond;

start2:

**bool**  b =  \_tmutex .timedWait(millsecond);

//完成处理了

**if** (finish())

**return**  **true** ;

**if** (!b)

**return**  **false** ;

millsecond = max(( int64\_t )0, m - ( TC\_Common ::*now2ms*() - iNow));

**goto**  start2;

**return**  **false** ;

\_tmutex .timedWait(millsecond)方法唤醒。反复判断是否所有的工作是否完成：

**bool** **TC\_ThreadPool::****finish**()

**return**  \_startqueue .empty() &&  \_jobqueue .empty() &&  \_busthread . empty() &&  \_bAllDone ;

整体cpp实现如下：

TC\_ThreadPool :: KeyInitialize *TC\_ThreadPool::g\_key\_initialize* ;

pthread\_key\_t *TC\_ThreadPool::g\_key* ;

**void** **TC\_ThreadPool::destructor**( **void**  \*p)

ThreadData  \*ttd = ( ThreadData \*)p;

**if** (ttd)

**delete**  ttd;

**void** **TC\_ThreadPool::exit**()

TC\_ThreadPool :: ThreadData  \*p = *getThreadData*();

**if** (p)

**delete**  p;

**int**  ret = pthread\_setspecific( *g\_key* , NULL );

**if** (ret != 0)

**throw**  TC\_ThreadPool\_Exception ( "\[TC\_ThreadPool::setThreadData\] pthread\_setspecific error" , ret);

\_jobqueue .clear();

**void** **TC\_ThreadPool::setThreadData**( TC\_ThreadPool :: ThreadData  \*p)

TC\_ThreadPool :: ThreadData  \*pOld = *getThreadData*();

**if** (pOld != NULL && pOld != p)

**delete**  pOld;

**int**  ret = pthread\_setspecific( *g\_key* , ( **void**  \*)p);

**if** (ret != 0)

**throw**  TC\_ThreadPool\_Exception ( "\[TC\_ThreadPool::setThreadData\] pthread\_setspecific error" , ret);

TC\_ThreadPool :: ThreadData \* **TC\_ThreadPool::getThreadData** ()

**return**  ( ThreadData  \*)pthread\_getspecific( *g\_key* );

**void** **TC\_ThreadPool::setThreadData**( pthread\_key\_t pkey,  ThreadData  \*p)

TC\_ThreadPool :: ThreadData  \*pOld = getThreadData(pkey);

**if** (pOld != NULL && pOld != p)

**delete**  pOld;

**int**  ret = pthread\_setspecific(pkey, ( **void**  \*)p);

**if** (ret != 0)

**throw**  TC\_ThreadPool\_Exception ( "\[TC\_ThreadPool::setThreadData\] pthread\_setspecific error" , ret);

TC\_ThreadPool :: ThreadData \* **TC\_ThreadPool::getThreadData**(pthread\_key\_t pkey)

**return**  ( ThreadData  \*)pthread\_getspecific(pkey);

**TC\_ThreadPool::TC\_ThreadPool**()

:  \_bAllDone ( **true** )

**TC\_ThreadPool::~TC\_ThreadPool**()

stop();

clear();

**void** **TC\_ThreadPool::clear**()

std::vector< ThreadWorker \*>::iterator it =  \_jobthread . begin();

**while** (it !=  \_jobthread .end())

**delete**  (\*it);

\++it;

\_jobthread .clear();

\_busthread .clear();

**void** **TC\_ThreadPool::init**( size\_t num)

stop();

Lock  sync(\* **this** );

clear();

**for** (size\_t i = 0; i < num; i++)

\_jobthread .push\_back( **new**  ThreadWorker( **this** ));

**void** **TC\_ThreadPool::stop**()

Lock  sync(\* **this** );

std::vector< ThreadWorker \*>::iterator it =  \_jobthread . begin();

**while** (it !=  \_jobthread .end())

**if**  ((\*it)->isAlive())

(\*it)-> terminate();

(\*it)-> getThreadControl().join ();

\++it;

\_bAllDone  =  **true** ;

**void** **TC\_ThreadPool::start**()

Lock  sync(\* **this** );

std::vector< ThreadWorker \*>::iterator it =  \_jobthread . begin();

**while** (it !=  \_jobthread .end())

(\*it)->start();

\++it;

\_bAllDone  =  **false** ;

**bool** **TC\_ThreadPool::****finish**()

**return**  \_startqueue .empty() &&  \_jobqueue .empty() &&  \_busthread . empty() &&  \_bAllDone ;

**bool** **TC\_ThreadPool::waitForAllDone**( **int**  millsecond)

Lock  sync( \_tmutex );

start1:

//任务队列和繁忙线程都是空的

**if**  (finish ())

**return**  **true** ;

//永远等待

**if** (millsecond < 0)

\_tmutex .timedWait(1000);

**goto**  start1;

int64\_t  iNow =  TC\_Common ::*now2ms*();

**int**  m = millsecond;

start2:

**bool**  b =  \_tmutex .timedWait(millsecond);

//完成处理了

**if** (finish ())

**return**  **true** ;

**if** (!b)

**return**  **false** ;

millsecond = max(( int64\_t )0, m - ( TC\_Common ::*now2ms*() - iNow));

**goto**  start2;

**return**  **false** ;

TC\_FunctorWrapperInterface \***TC\_ThreadPool::get**( ThreadWorker  \*ptw)

TC\_FunctorWrapperInterface  \*pFunctorWrapper = NULL;

**if** (! \_jobqueue .pop\_front(pFunctorWrapper, 1000))

**return** NULL;

Lock  sync( \_tmutex );

\_busthread . insert(ptw);

**return**  pFunctorWrapper;

TC\_FunctorWrapperInterface \***TC\_ThreadPool::get**()

TC\_FunctorWrapperInterface  \*pFunctorWrapper = NULL;

**if** (! \_startqueue .pop\_front(pFunctorWrapper))

**return** NULL;

**return**  pFunctorWrapper;

**void** **TC\_ThreadPool::idle**( ThreadWorker  \*ptw)

Lock  sync( \_tmutex );

\_busthread .erase(ptw);

//无繁忙线程, 通知等待在线程池结束的线程醒过来

**if** ( \_busthread .empty())

\_bAllDone  =  **true** ;

\_tmutex .notifyAll();

**void** **TC\_ThreadPool::notifyT**()

\_jobqueue .notifyT();

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

C++ 线程池的封装实现

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 C++实现一个线程池

相关 c++实现简单的线程池

相关 C++封装POSIX 线程库（三）线程的封装

相关 C语言实现的线程池

相关 C++ 线程池的封装实现

相关 C++封装POSIX 线程库（六）线程池

相关 C++ 线程的封装

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

相关 C++实现一个线程池

相关 Android线程池封装库