[C++][线程池][完整实现] 转：线程池原理及创建（C++实现）

文章的主要框架是参考这篇文档的，http://jacky-dai.iteye.com/blog/1090285,

关于作者 

张中庆，西安交通大学软件所，在读硕士，目前研究方向为分布式网络与移动中间件，对Linux极其爱好，可以通过flydish1234@sina.com.cn与我联系

其实看了好多类似的，都是没有完整的实现，自己花了点时间把这个程序梳理了一下，写了个测试程序，目前可以跑通。

#include <stdio.h>
#include <stdlib.h>
#include <vector>
#include <algorithm>
#include <assert.h>
#include <Windows.h>
#include <functional>
#include <process.h>

using namespace std;
class CThread;

//锁的基类
class CLockObject
{
public:
virtual BOOL Lock() = 0;
virtual BOOL UnLock() = 0;
};

//任务基类，所有要执行的任务都继承这个类
class CJob
{
private:
int m_JobNo;//任务ID 用来调试是否绑定特定线程
char* m_JobName; //任务名字，用来调试是否绑定特定线程
CThread* m_pWorkThread; //The thread associated with the job
public:
CJob();
virtual ~CJob();
CThread *GetWorkThread(void); //获取工作线程
void SetWorkThread(CThread* pWorkThread);//设置工作线程
virtual void Execute(void* ptr) = 0; //执行函数
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);
};

void CJob::SetJobName(char* jobname)
{
if(NULL !=m_JobName)
{
free(m_JobName);
m_JobName = NULL;
}

if(NULL !=jobname)
{
m_JobName = (char*)malloc(strlen(jobname)+1);
strcpy(m_JobName,jobname);
}
}

CThread* CJob::GetWorkThread(void)
{
return m_pWorkThread;
}

void CJob::SetWorkThread(CThread *pWorkThread)
{
m_pWorkThread = pWorkThread;
}

CJob::CJob(void) :m_pWorkThread(NULL),m_JobName(NULL), m_JobNo(0)
{

}

CJob::~CJob()
{
if (NULL != m_JobName)
{
free(m_JobName);
m_JobName = NULL;
}
}

//线程状态
typedef enum _ThreadState
{
THREAD_RUNNING = 0x0, //运行
THREAD_IDLE = 0x1,//空闲
THREAD_EXIT = 0X2,//退出
}ThreadState;

//线程基类
class CThread
{
private:
int m_ErrorCode; //错误码
unsigned long m_ThreadID; //线程ID
char* m_ThreadName; //线程名字
ThreadState m_ThreadState; //线程状态
HANDLE m_hthreadHandle; //线程句柄
bool m_IsExit;//是否退出
protected:
static unsigned __stdcall ThreadFunction(void*); //start调用此函数，此函数再调用run函数，执行实际的任务
public:
CThread();
virtual ~CThread();

virtual void Run() = 0;

//设置线程状态
void SetThreadState(ThreadState state);
//获取线程状态
ThreadState GetThreadState();

//Start to execute the thread
bool Start();

//获取线程ID
int GetThreadID(void);

//设置错误码
void SetErrorCode(int errorCode);
//获取错误码
int GetLastError(void);

//设置线程名字
void SetThreadName(char* threadName);
//获取线程名字
char* GetThreadName();

//设置线程优先级
bool SetPriority(int priority);
//获取线程优先级
int GetPriority(void);

bool Terminate(void);
HANDLE GetThreadHandle();
void SetThreadHandle(HANDLE hdl);
void SetExitFlag(bool bExit);
bool GetExitFlag();
bool NeedExit();
};

bool CThread::NeedExit()
{
return m_IsExit;
}

void CThread::SetExitFlag(bool bExit)
{
m_IsExit = bExit;
}

bool CThread::GetExitFlag()
{
return m_IsExit;
}

bool CThread::Terminate(void)
{
_endthreadex(0);
return TRUE;
}

HANDLE CThread::GetThreadHandle()
{
return m_hthreadHandle;
}

void CThread::SetThreadHandle(HANDLE hdl)
{
m_hthreadHandle = hdl;
}

void CThread::SetErrorCode(int errorCode)
{
m_ErrorCode = errorCode;
}

int CThread::GetLastError(void)
{
return m_ErrorCode;
}

CThread::CThread()
{
m_IsExit = FALSE;
}

CThread::~CThread()
{

}

void CThread::SetThreadState(ThreadState state)
{
m_ThreadState = state;
}

ThreadState CThread::GetThreadState()
{
return m_ThreadState;
}

//Start to execute the thread
bool CThread::Start()
{
unsigned threadID;
HANDLE hThread = (HANDLE)_beginthreadex(NULL, 0, ThreadFunction, this, 0, &threadID);
this->m_ThreadID = threadID;
this->SetThreadHandle(hThread);
return true;
}

unsigned __stdcall CThread::ThreadFunction(void* pArg)
{
CThread* pThread = (CThread*)pArg;
pThread->Run();
return TRUE;
}

int CThread::GetThreadID(void)
{
return m_ThreadID;
}

void CThread::SetThreadName(char* threadName)
{
strncpy(m_ThreadName, threadName, strlen(threadName));
}

char* CThread::GetThreadName()
{
return m_ThreadName;
}

//线程互斥锁
class CThreadMutex: public CLockObject
{
private:
CRITICAL_SECTION m_CritSec;//临界区
public:
CThreadMutex();
~CThreadMutex();
BOOL Lock();//加锁，阻塞式
BOOL UnLock();//解锁
BOOL TryLock();//加锁，非阻塞式
};

CThreadMutex::CThreadMutex()
{
#if (_WIN32_WINNT >= 0x0403)
//使用 InitializeCriticalSectionAndSpinCount 可以提高性能
::InitializeCriticalSectionAndSpinCount(&m_CritSec,4000);
#else
::InitializeCriticalSection(&m_CritSec);
#endif
}

CThreadMutex::~CThreadMutex()
{
::DeleteCriticalSection(&m_CritSec);
}

BOOL CThreadMutex::Lock()
{
::EnterCriticalSection(&m_CritSec);
return TRUE;
}

BOOL CThreadMutex::UnLock()
{
::LeaveCriticalSection(&m_CritSec);
return TRUE;
}

BOOL CThreadMutex::TryLock()
{
BOOL bRet = TryEnterCriticalSection(&m_CritSec);
return bRet;
}

//条件变量
class CThreadCondition
{
private:
HANDLE m_phEvent; //句柄
public:
CThreadCondition();
~CThreadCondition();
void Wait();
void Signal();
};

CThreadCondition::CThreadCondition()
{
//第二个参数 bManualReset FALSE the system automatically resets the state to nonsignaled
//If this parameter is TRUE, the function creates a manual-reset event object
//第三个参数 bInitialState FALSE it is nonsignaled
m_phEvent = ::CreateEvent(NULL, TRUE, FALSE, NULL);
}

CThreadCondition::~CThreadCondition()
{
if (NULL != m_phEvent)
{
::CloseHandle((m_phEvent));
}
}

void CThreadCondition::Wait()
{
//If dwMilliseconds is INFINITE, the function will return only when the object is signaled.
WaitForSingleObject(m_phEvent, INFINITE);
ResetEvent(m_phEvent);
}

void CThreadCondition::Signal()
{
//Sets the specified event object to the signaled state
SetEvent(m_phEvent);
}

//线程池类，主要负责调度线程，创建线程，删除线程
class CThreadPool
{
friend class CWorkerThread;
private:
unsigned int m_nMaxNum; //当前线程池中所允许并发存在的线程的最大数目
unsigned int m_nAvailLow; //当前线程池中所允许存在的空闲线程的最小数目
//如果空闲数目低于该值，表明负载可能过重，此时有必要增加空闲线程池的数目
//实现中我们总是将线程调整为m_InitNum个
unsigned int m_nAvailHigh;//当前线程池中所允许的空闲的线程的最大数目，
//如果空闲数目高于该值，表明当前负载可能较轻，此时将删除多余的空闲线程，删除后调整数也为m_InitNum个
unsigned int m_nCurIdleThreadsNum;//当前线程池中实际存在的线程的个数，其值介于m_nAvailHigh和m_nAvailLow之间
//如果线程的个数始终维持在m_nAvailLow和m_nAvailHigh之间，则线程既不需要创建，也不需要删除，保持平衡状态
unsigned int m_nInitThreadsNum;//初始创建时线程池中的线程的个数

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_ThreadNumMutex;//变量锁， 目前用在m_nCurIdleThreadsNum修改上面

CThreadCondition m_BusyCond; //m_BusyCond is used to sync busy thread list
CThreadCondition m_IdleCond; //m_IdleCond is used to sync idle thread list
CThreadCondition m_MaxNumCond;//m_MaxNumCond is used to sync m_nCurIdleThreadsNum

vector<CWorkerThread*> m_vecAllThreads;//所有创建出来的线程集合
vector<CWorkerThread*> m_vecBusyThreads;//忙碌线程队列，随着负载的多少会改变
vector<CWorkerThread*> m_vecIdleThreads;//空闲线程队列,随着负的多少会改变

public:
void SetMaxNum(int maxnum){m_nMaxNum = maxnum;} //设置线程池运行的最大线程数
int GetMaxNum(void){return m_nMaxNum;}
void SetAvailLowNum(int minnum){m_nAvailLow = minnum;} //设置最少空闲线程数
int GetAvailLowNum(void){return m_nAvailLow;}
void SetAvailHighNum(int highnum){m_nAvailHigh = highnum;} //设置最多空闲线程数
int GetAvailHighNum(void){return m_nAvailHigh;}
int GetCurIdleThreadsNum(void){return m_nCurIdleThreadsNum;} //获取当前空闲线程个数
int GetAllThreadsNum(void){return m_vecAllThreads.size();} //获取所有线程个数
int GetBusyThreadsNum(void){return m_vecBusyThreads.size();} //获取忙碌空闲线程个数
void SetInitNum(int initnum){m_nInitThreadsNum = initnum;}
int GetInitNum(void){return m_nInitThreadsNum;}
CThreadPool();
~CThreadPool();
CThreadPool(int initnum);
void TerminateAll();
void Run(CJob* job,void* jobdata);
};

//真正的工作线程，执行操作的线程
class CWorkerThread : public CThread
{
private:
CThreadPool* m_pThreadPool;//线程池
CJob* m_pJob;//任务
void* m_pJobData;//任务参数
CThreadMutex m_VarMutex;//
public:
CThreadCondition m_JobAddCond; //有新的任务时触发条件变量,每个线程一个条件变量，可以指定线程去执行任务
CThreadMutex m_WorkMutex;//
CWorkerThread();
virtual ~CWorkerThread();
void Run();
void AddNewJob(CJob* job,void* jobdata);
CJob* GetJob(void){return m_pJob;}
void SetThreadPool(CThreadPool* thrpool);
CThreadPool* GetThreadPool(void){return m_pThreadPool;}
void Terminate(void);
};

void CWorkerThread::Terminate(void)
{
//工作线程再处理任务结束才会解锁，这个时候再去退出线程，避免打断线程处理任务。
m_WorkMutex.Lock();
SetExitFlag(TRUE);
//工作为假 代表要求线程退出
m_pJob = NULL;
m_pJobData = NULL;
printf("thread [%d] ready to exit\n", GetThreadID());
m_JobAddCond.Signal();
m_WorkMutex.UnLock();
WaitForSingleObject(GetThreadHandle(), INFINITE);
CloseHandle(GetThreadHandle());
}

CWorkerThread::CWorkerThread()
{
m_pJobData = NULL;
m_pJob = NULL;
m_pThreadPool = NULL;
}

CWorkerThread::~CWorkerThread()
{
if (NULL != m_pJob) {delete m_pJob; m_pJob = NULL;}
if (NULL != m_pThreadPool) {delete m_pThreadPool; m_pThreadPool = NULL;}
}

void CWorkerThread::Run()
{
printf("Enter CWorkerThread::Run\n");
SetThreadState(THREAD_RUNNING);
for(;;)
{
//当前线程不退出才需要等待任务的到来
while ((NULL == m_pJob) && !NeedExit())
{
printf("thread [%d] wait for job \n", GetThreadID());
m_JobAddCond.Wait();
}

if (NULL == m_pJob)
{
printf("thread [%d] exitFlag [%d]\n", GetThreadID(), NeedExit());
if (NeedExit())
{
break;//不再等待任务，退出线程
}
else
{
//任务为NULL 但不是线程退出，跳过这个任务
printf("m_pJob [%p] exitFlag [%d]\n", m_pJob, NeedExit());
continue;
}
}

m_WorkMutex.Lock();
printf("thread [%d] accept the job [%d]\n", GetThreadID(), m_pJob->GetJobNo());
//真正执行任务的地方
m_pJob->Execute(m_pJobData);
m_pJob->SetWorkThread(NULL);
m_pJob = NULL;
m_pJobData = NULL;
m_pThreadPool->MoveToIdleList(this);
SetThreadState(THREAD_IDLE);
if(m_pThreadPool->m_vecIdleThreads.size() > m_pThreadPool->GetAvailHighNum())
{
m_pThreadPool->DeleteIdleThread(m_pThreadPool->m_vecIdleThreads.size() - m_pThreadPool->GetInitNum());
}
m_WorkMutex.UnLock();
}
printf("thread [%d] exit\n", GetThreadID());
}

void CWorkerThread::AddNewJob(CJob* pJob,void* jobdata)
{
assert(NULL != pJob);
m_VarMutex.Lock();
m_pJob = pJob;
m_pJobData = jobdata;
pJob->SetWorkThread(this);
m_VarMutex.UnLock();
printf("job [%d] add to the pool\n",m_pJob->GetJobNo());
m_JobAddCond.Signal();
}

void CWorkerThread::SetThreadPool(CThreadPool* thrpool)
{
m_VarMutex.Lock();
m_pThreadPool = thrpool;
m_VarMutex.UnLock();
}

CThreadPool::CThreadPool()
{
m_nMaxNum = 50;
m_nAvailLow = 5;
m_nInitThreadsNum = 10;
m_nCurIdleThreadsNum = 10;
m_nAvailHigh = 20;
m_vecBusyThreads.clear();
m_vecIdleThreads.clear();
int i;
for (i=0; i<m_nInitThreadsNum; ++i)
{
CWorkerThread* pNewWorkThread = new CWorkerThread;
pNewWorkThread->SetThreadPool(this);
AppendToIdleList(pNewWorkThread);
pNewWorkThread->Start();
}
}

CThreadPool::CThreadPool(int initnum)
{
m_nMaxNum = 30;
m_nAvailLow = (initnum-10>0)?(initnum-10):3;
m_nInitThreadsNum = m_nCurIdleThreadsNum = initnum ;
m_nAvailHigh = initnum+10;
m_vecAllThreads.clear();
m_vecBusyThreads.clear();
m_vecIdleThreads.clear();
int i;
for (i=0; i<m_nInitThreadsNum; ++i)
{
CWorkerThread* pNewWorkThread = new CWorkerThread;
pNewWorkThread->SetThreadPool(this);
AppendToIdleList(pNewWorkThread);
pNewWorkThread->Start();
}
printf("CThreadPool::CThreadPool: Create Thread [%d] success\n", m_nInitThreadsNum);
}

CThreadPool::~CThreadPool()
{
TerminateAll();
}

void CThreadPool::TerminateAll()
{
int i;
for (i=0; i<m_vecAllThreads.size(); ++i)
{
CWorkerThread* pWorkThread = m_vecAllThreads[i];
pWorkThread->Terminate();
}
}

//获取空闲的线程
CWorkerThread* CThreadPool::GetIdleThread(void)
{
while (0 == m_vecIdleThreads.size())
{
printf("no idle threads, must wait\n");
m_IdleCond.Wait();
}
m_IdleMutex.Lock();
if (0 < m_vecIdleThreads.size())
{
CWorkerThread* pIdleThread = (CWorkerThread*)m_vecIdleThreads.front();
printf("get idle thread %d \n", pIdleThread->GetThreadID());
m_IdleMutex.UnLock();
return pIdleThread;
}
m_IdleMutex.UnLock();
printf("warning: no idle threads return\n");
return NULL;
}

//add an idle thread to idle list
void CThreadPool::AppendToIdleList(CWorkerThread* jobthread)
{
m_IdleMutex.Lock();
m_vecIdleThreads.push_back(jobthread);
m_vecAllThreads.push_back(jobthread);
m_IdleMutex.UnLock();
}

//move and idle thread to busy thread
void CThreadPool::MoveToBusyList(CWorkerThread* idlethread)
{
m_BusyMutex.Lock();
m_vecBusyThreads.push_back(idlethread);
m_nCurIdleThreadsNum--;
m_BusyMutex.UnLock();

m_IdleMutex.Lock();
vector<CWorkerThread*>::iterator pos;
pos = find(m_vecIdleThreads.begin(),m_vecIdleThreads.end(),idlethread);
if(pos != m_vecIdleThreads.end())
{
m_vecIdleThreads.erase(pos);
}
m_IdleMutex.UnLock();
}

void CThreadPool::MoveToIdleList(CWorkerThread* busythread)
{
m_IdleMutex.Lock();
m_vecIdleThreads.push_back(busythread);
m_nCurIdleThreadsNum++;
m_IdleMutex.UnLock();

m_BusyMutex.Lock();
vector<CWorkerThread*>::iterator pos;
pos = find(m_vecBusyThreads.begin(),m_vecBusyThreads.end(),busythread);
if(pos!=m_vecBusyThreads.end())
m_vecBusyThreads.erase(pos);
m_BusyMutex.UnLock();
m_IdleCond.Signal();
m_MaxNumCond.Signal();
}

//create num idle thread and put them to idlelist
void CThreadPool::CreateIdleThread(int num)
{
int i;
for (i=0;i<num;i++)
{
CWorkerThread* pWorkThread = new CWorkerThread();
pWorkThread->SetThreadPool(this);
AppendToIdleList(pWorkThread);
m_ThreadNumMutex.Lock();
m_nCurIdleThreadsNum++;
m_ThreadNumMutex.UnLock();
pWorkThread->Start();//begin the thread,the thread wait for job
}
}

void CThreadPool::DeleteIdleThread(int num)
{
printf("Enter into CThreadPool::DeleteIdleThread\n");
m_IdleMutex.Lock();
printf("Delete Num is %dn",num);
int i;
for(i=0;i<num;i++)
{
CWorkerThread* thr;
if(m_vecIdleThreads.size() > 0 )
{
thr = (CWorkerThread*)m_vecIdleThreads.front();
printf("Get Idle thread %dn",thr->GetThreadID());
}
else
{
printf("no idle thread, no need to delete thread\n");
break;
}

vector<CWorkerThread*>::iterator pos;
pos = find(m_vecIdleThreads.begin(),m_vecIdleThreads.end(),thr);
if(pos!=m_vecIdleThreads.end())
m_vecIdleThreads.erase(pos);
m_nCurIdleThreadsNum--;
printf("The idle thread available num:%d n",m_nCurIdleThreadsNum);
printf("The idlelist num:%d n",m_vecIdleThreads.size());
}
m_IdleMutex.UnLock();
}

void CThreadPool::Run(CJob* job, void* jobdata)
{
assert(NULL != job);
//if the busy thread num adds to m_nMaxNum,so we should wait
if(m_nMaxNum <= GetBusyThreadsNum())
{
printf("busy threads beyond the max threads number in the pool, must wait for idle threads\n");
m_MaxNumCond.Wait();
}

//负载过重，空闲线程少，需要创建新的线程, 使其数目达到m_InitNum
if(m_vecIdleThreads.size() < m_nAvailLow)
{
if(GetAllThreadsNum()+m_nInitThreadsNum-m_vecIdleThreads.size() < m_nMaxNum )
{
//当前有m_vecIdleThreads.size()空闲线程， 另外再创建m_nInitThreadsNum - m_vecIdleThreads.size()， 当前总的空闲线程为m_nInitThreadsNum
CreateIdleThread(m_nInitThreadsNum - m_vecIdleThreads.size());
}
else
{
CreateIdleThread(m_nMaxNum - GetAllThreadsNum());
}
}

CWorkerThread* pWorkthread = GetIdleThread();
if(NULL != pWorkthread)
{
pWorkthread->m_WorkMutex.Lock();
MoveToBusyList(pWorkthread);
pWorkthread->SetThreadPool(this);
job->SetWorkThread(pWorkthread);
printf("Job [%d] bind to thread [%d] \n", job->GetJobNo(), pWorkthread->GetThreadID());
pWorkthread->AddNewJob(job, jobdata);
pWorkthread->m_WorkMutex.UnLock();
}
else
{
printf("impossible to going here\n");
}
}

//线程管理类
class CThreadManage
{
public:
CThreadManage();
CThreadManage(int num);
virtual ~CThreadManage();
void Run(CJob* pjob, void* pJobData);//运行任务
void TerminateAll(); //停止所有的线程
private:
int m_nNumOfThread; //初始时允许创建的最大的线程个数
CThreadPool* m_pPool;//实际的线程池
};

CThreadManage::CThreadManage()
{
m_nNumOfThread = 10;
m_pPool = new CThreadPool(m_nNumOfThread);
}

CThreadManage::CThreadManage(int num)
{
m_nNumOfThread = num;
m_pPool = new CThreadPool(m_nNumOfThread);
}

CThreadManage::~CThreadManage()
{
if (NULL != m_pPool)
{
delete m_pPool;
m_pPool = NULL;
}
}

void CThreadManage::Run(CJob* pjob, void* pJobData)
{
m_pPool->Run(pjob, pJobData);
}

void CThreadManage::TerminateAll()
{
m_pPool->TerminateAll();
}

class myjob : public CJob
{
public:
myjob(){}
myjob(int i){SetJobNo(i);}
~myjob(){}
void Execute(void* jobdata) {
printf("The Job comes from CXJOB\n");
::Sleep(2);
}
};

#if 0
int main()
{
CThreadManage* manage = new CThreadManage(50);
for(int i=0;i<1000;i++)
{
myjob* job = new myjob(i);
manage->Run(job, NULL);
}
::Sleep(2);

myjob* job = new myjob();
manage->Run(job, NULL);
manage->TerminateAll();
return 0;
}
#endif