【线程池C++】

大家好，欢迎来到IT知识分享网。

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

什么是线程

线程（英语：thread）是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。在Unix System V及SunOS中也被称为轻量进程（lightweight processes），但轻量进程更多指内核线程（kernel thread），而把用户线程（user thread）称为线程。

什么是线程池及为什么需要线程池

线程池（英语：thread pool）：一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。 例如，线程数一般取cpu数量+2比较合适，线程数过多会导致额外的线程切换开销。

目前的大多数网络服务器，包括Web服务器、Email服务器以及数据库服务器等都具有一个共同点，就是单位时间内必须处理数目巨大的连接请求，但处理时间却相对较短。

传统多线程方案中我们采用的服务器模型则是一旦接受到请求之后，即创建一个新的线程，由该线程执行任务。任务执行完毕后，线程退出，这就是是“即时创建，即时销毁”的策略。尽管与创建进程相比，创建线程的时间已经大大的缩短，但是如果提交给线程的任务是执行时间较短，而且执行次数极其频繁，那么服务器将处于不停的创建线程，销毁线程的状态。

除此之外，线程池能够减少创建的线程个数。通常线程池所允许的并发线程是有上界的，如果同时需要并发的线程数超过上界，那么一部分线程将会等待。而传统方案中，如果同时请求数目为2000，那么最坏情况下，系统可能需要产生2000个线程。尽管这不是一个很大的数目，但是也有部分机器可能达不到这种要求。

使用线程池的好处

降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

提高响应速度。当任务到达时，任务可以不需要的等到线程创建就能立即执行。

提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统 的稳定性，使用线程池可以进行统一的分配，调优和监控。

构建线程池框架

一般线程池都必须具备下面几个组成部分：

• 线程池管理器:用于创建并管理线程池
• 工作线程: 线程池中实际执行的线程
• 任务接口: 尽管线程池大多数情况下是用来支持网络服务器，但是我们将线程执行的任务抽象出来，形成任务接口，从而是的线程池与具体的任务无关。
• 任务队列: 线程池的概念具体到实现则可能是队列，链表之类的数据结构，其中保存执行线程。

我们实现的通用线程池框架由五个重要部分组成CThreadManage，CThreadPool，CThread，CJob，CWorkerThread，除此之外框架中还包括线程同步使用的类CThreadMutex和CCondition。

CThreadManage直接跟客户端打交道，其接受需要创建的线程初始个数，并接受客户端提交的任务。这儿的任务是具体的非抽象的任务。CThreadManage的内部实际上调用的都是CThreadPool的相关操作。CThreadPool创建具体的线程，并把客户端提交的任务分发给CWorkerThread，CWorkerThread实际执行具体的任务。

理解系统组件

class CThreadManage { private: CThreadPool* m_Pool; int m_NumOfThread; protected: public: CThreadManage(); CThreadManage(int num); virtual ~CThreadManage(); void SetParallelNum(int num); void Run(CJob* job,void* jobdata); void TerminateAll(void); };

其中m_Pool指向实际的线程池；m_NumOfThread是初始创建时候允许创建的并发的线程个数。另外Run和TerminateAll方法也非常简单，只是简单的调用CThreadPool的一些相关方法而已。其具体的实现如下:

CThreadManage::CThreadManage() { m_NumOfThread = 10; m_Pool = new CThreadPool(m_NumOfThread); } CThreadManage::CThreadManage(int num) { m_NumOfThread = num; m_Pool = new CThreadPool(m_NumOfThread); } CThreadManage::~CThreadManage() { if(NULL != m_Pool) delete m_Pool; } void CThreadManage::SetParallelNum(int num) { m_NumOfThread = num; } void CThreadManage::Run(CJob* job,void* jobdata) { m_Pool->Run(job,jobdata); } void CThreadManage::TerminateAll(void) { m_Pool->TerminateAll(); }

CThread
CThread 类实现了对Linux中线程操作的封装，它是所有线程的基类，也是一个抽象类，提供了一个抽象接口Run，所有的CThread都必须实现该Run方法。CThread的定义如下所示：

class CThread { private: int m_ErrCode; Semaphore m_ThreadSemaphore; //the inner semaphore, which is used to realize unsigned long m_ThreadID; bool m_Detach; //The thread is detached bool m_CreateSuspended; //if suspend after creating char* m_ThreadName; ThreadState m_ThreadState; //the state of the thread protected: void SetErrcode(int errcode){m_ErrCode = errcode;} static void* ThreadFunction(void*); public: CThread(); CThread(bool createsuspended,bool detach); virtual ~CThread(); virtual void Run(void) = 0; void SetThreadState(ThreadState state){m_ThreadState = state;} bool Terminate(void); //Terminate the threa bool Start(void); //Start to execute the thread void Exit(void); bool Wakeup(void); ThreadState GetThreadState(void){return m_ThreadState;} int GetLastError(void){return m_ErrCode;} void SetThreadName(char* thrname){strcpy(m_ThreadName,thrname);} char* GetThreadName(void){return m_ThreadName;} int GetThreadID(void){return m_ThreadID;} bool SetPriority(int priority); int GetPriority(void); int GetConcurrency(void); void SetConcurrency(int num); bool Detach(void); bool Join(void); bool Yield(void); int Self(void); };

class CThreadPool { friend class CWorkerThread; private: unsigned int m_MaxNum; //the max thread num that can create at the same time unsigned int m_AvailLow; //The min num of idle thread that shoule kept unsigned int m_AvailHigh; //The max num of idle thread that kept at the same time unsigned int m_AvailNum; //the normal thread num of idle num; unsigned int m_InitNum; //Normal thread num; protected: CWorkerThread* GetIdleThread(void); void AppendToIdleList(CWorkerThread* jobthread); void MoveToBusyList(CWorkerThread* idlethread); void MoveToIdleList(CWorkerThread* busythread); void DeleteIdleThread(int num); void CreateIdleThread(int num); public: CThreadMutex m_BusyMutex; //when visit busy list,use m_BusyMutex to lock and unlock CThreadMutex m_IdleMutex; //when visit idle list,use m_IdleMutex to lock and unlock CThreadMutex m_JobMutex; //when visit job list,use m_JobMutex to lock and unlock CThreadMutex m_VarMutex; CCondition m_BusyCond; //m_BusyCond is used to sync busy thread list CCondition m_IdleCond; //m_IdleCond is used to sync idle thread list CCondition m_IdleJobCond; //m_JobCond is used to sync job list CCondition m_MaxNumCond; vector<CWorkerThread*> m_ThreadList; vector<CWorkerThread*> m_BusyList; //Thread List vector<CWorkerThread*> m_IdleList; //Idle List CThreadPool(); CThreadPool(int initnum); virtual ~CThreadPool(); void SetMaxNum(int maxnum){m_MaxNum = maxnum;} int GetMaxNum(void){return m_MaxNum;} void SetAvailLowNum(int minnum){m_AvailLow = minnum;} int GetAvailLowNum(void){return m_AvailLow;} void SetAvailHighNum(int highnum){m_AvailHigh = highnum;} int GetAvailHighNum(void){return m_AvailHigh;} int GetActualAvailNum(void){return m_AvailNum;} int GetAllNum(void){return m_ThreadList.size();} int GetBusyNum(void){return m_BusyList.size();} void SetInitNum(int initnum){m_InitNum = initnum;} int GetInitNum(void){return m_InitNum;} void TerminateAll(void); void Run(CJob* job,void* jobdata); }; CWorkerThread* CThreadPool::GetIdleThread(void) { while(m_IdleList.size() ==0 ) m_IdleCond.Wait(); m_IdleMutex.Lock(); if(m_IdleList.size() > 0 ) { CWorkerThread* thr = (CWorkerThread*)m_IdleList.front(); printf("Get Idle thread %d\n",thr->GetThreadID()); m_IdleMutex.Unlock(); return thr; } m_IdleMutex.Unlock(); return NULL; } //create num idle thread and put them to idlelist void CThreadPool::CreateIdleThread(int num) { for(int i=0;i<num;i++){ CWorkerThread* thr = new CWorkerThread(); thr->SetThreadPool(this); AppendToIdleList(thr); m_VarMutex.Lock(); m_AvailNum++; m_VarMutex.Unlock(); thr->Start(); //begin the thread,the thread wait for job } } void CThreadPool::Run(CJob* job,void* jobdata) { assert(job!=NULL); //if the busy thread num adds to m_MaxNum,so we should wait if(GetBusyNum() == m_MaxNum) m_MaxNumCond.Wait(); if(m_IdleList.size()<m_AvailLow) { if(GetAllNum()+m_InitNum-m_IdleList.size() < m_MaxNum ) CreateIdleThread(m_InitNum-m_IdleList.size()); else CreateIdleThread(m_MaxNum-GetAllNum()); } CWorkerThread* idlethr = GetIdleThread(); if(idlethr !=NULL) { idlethr->m_WorkMutex.Lock(); MoveToBusyList(idlethr); idlethr->SetThreadPool(this); job->SetWorkThread(idlethr); printf("Job is set to thread %d \n",idlethr->GetThreadID()); idlethr->SetJob(job,jobdata); } }

在CThreadPool中存在两个链表，一个是空闲链表，一个是忙碌链表。Idle链表中存放所有的空闲进程，当线程执行任务时候，其状态变为忙碌状态，同时从空闲链表中删除，并移至忙碌链表中。在CThreadPool的构造函数中，我们将执行下面的代码:

for(int i=0;i<m_InitNum;i++) { CWorkerThread* thr = new CWorkerThread(); AppendToIdleList(thr); thr->SetThreadPool(this); thr->Start(); //begin the thread,the thread wait for job }

m_InitNum：处世创建时线程池中的线程的个数。

m_MaxNum:当前线程池中所允许并发存在的线程的最大数目。

m_AvailLow:当前线程池中所允许存在的空闲线程的最小数目，如果空闲数目低于该值，表明负载可能过重，此时有必要增加空闲线程池的数目。实现中我们总是将线程调整为m_InitNum个。

m_AvailHigh：当前线程池中所允许的空闲的线程的最大数目，如果空闲数目高于该值，表明当前负载可能较轻，此时将删除多余的空闲线程，删除后调整数也为m_InitNum个。

if(GetAllNum()+m_InitNum-m_IdleList.size() < m_MaxNum ) CreateIdleThread(m_InitNum-m_IdleList.size()); else CreateIdleThread(m_MaxNum-GetAllNum());

class CJob { private: int m_JobNo; //The num was assigned to the job char* m_JobName; //The job name CThread *m_pWorkThread; //The thread associated with the job public: CJob( void ); virtual ~CJob(); int GetJobNo(void) const { return m_JobNo; } void SetJobNo(int jobno){ m_JobNo = jobno;} char* GetJobName(void) const { return m_JobName; } void SetJobName(char* jobname); CThread *GetWorkThread(void){ return m_pWorkThread; } void SetWorkThread ( CThread *pWorkThread ){ m_pWorkThread = pWorkThread; } virtual void Run ( void *ptr ) = 0; }; 

线程池使用示例
至此我们给出了一个简单的与具体任务无关的线程池框架。使用该框架非常的简单，我们所需要的做的就是派生CJob类，将需要完成的任务实现在Run方法中。然后将该Job交由CThreadManage去执行。下面我们给出一个简单的示例程序

class CXJob:public CJob { public: CXJob(){i=0;} ~CXJob(){} void Run(void* jobdata) { printf("The Job comes from CXJOB\n"); sleep(2); } }; class CYJob:public CJob { public: CYJob(){i=0;} ~CYJob(){} void Run(void* jobdata) { printf("The Job comes from CYJob\n"); } }; main() { CThreadManage* manage = new CThreadManage(10); for(int i=0;i<40;i++) { CXJob* job = new CXJob(); manage->Run(job,NULL); } sleep(2); CYJob* job = new CYJob(); manage->Run(job,NULL); manage->TerminateAll(); } 

CXJob和CYJob都是从Job类继承而来，其都实现了Run接口。CXJob只是简单的打印一句”The Job comes from CXJob”，CYJob也只打印”The Job comes from CYJob”，然后均休眠2秒钟。在主程序中我们初始创建10个工作线程。然后分别执行40次CXJob和一次CYJob。

C++ 线程池的封装实现
为了充分利用多核的优势，我们利用多线程来进行任务处理，但线程也同样不能滥用，会带来一下几个问题：
1）线程本身存在开销，系统必须为每个线程分配如栈，TLS（线程局部存储），寄存器等。
2）线程管理会给系统带来开销，context切换同样会给系统带来成本。
3）线程本身是可以重用的资源，不需要每次都进行初始化。

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

主要含有三个队列

1. 工作队列
2. 工作线程队列
3. 忙碌线程队列

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(); }

线程池适合场合

参考文章：

深入解析C++编程中线程池的使用_C 语言_脚本之家 (jb51.net)