线程池原理及创建(C++实现)
本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进行了高层次的抽象,使之与具体的执行任务无关。另外该线程池具有动态伸缩性,它能根据执行任务的轻重自动调整线程池中线程的数量。文章的最后,我们给出一个简单示例程序,通过该示例程序,我们会发现,通过该线程池框架执行多线程任务是多么的简单。
为什么需要线程池
目前的大多数网络服务器,包括Web服务器、Email服务器以及数据库服务器等都具有一个共同点,就是单位时间内必须处理数目巨大的连接请求,但处理时间却相对较短。
我们将传统方案中的线程执行过程分为三个过程:T1、T2、T3。
基于这种预创建技术,线程池将线程创建和销毁本身所带来的开销分摊到了各个具体的任务上,执行次数越多,每个任务所分担到的线程本身开销则越小,不过我们另外可能需要考虑进去线程之间同步所带来的开销。
构建线程池框架
任务接口::尽管线程池大多数情况下是用来支持网络服务器,但是我们将线程执行的任务抽象出来,形成任务接口,从而是的线程池与具体的任务无关。
任务队列:线程池的概念具体到实现则可能是队列,链表之类的数据结构,其中保存执行线程。
CJob是所有的任务的基类,其提供一个接口Run,所有的任务类都必须从该类继承,同时实现Run方法。该方法中实现具体的任务逻辑。
CThread是Linux中线程的包装,其封装了Linux线程最经常使用的属性和方法,它也是一个抽象类,是所有线程类的基类,具有一个接口Run。
CWorkerThread是实际被调度和执行的线程类,其从CThread继承而来,实现了CThread中的Run方法。
CThreadPool是线程池类,其负责保存线程,释放线程以及调度线程。
CThreadManage是线程池与用户的直接接口,其屏蔽了内部的具体实现。
CCondition则是条件变量的封装,用于线程之间的同步。
(若有了线程池框架,我们所需要的做的就是派生CJob类,将需要完成的任务实现在Run方法中。然后将该Job交由CThreadManage去执行)
理解系统组件
CThreadManage
CThreadManage的功能非常简单,其提供最简单的方法,其类定义如下:
void Run(CJob* job,void* jobdata);
CThreadManage::CThreadManage(){
m_Pool = new CThreadPool(m_NumOfThread);
CThreadManage::CThreadManage(int num){
m_Pool = new CThreadPool(m_NumOfThread);
CThreadManage::~CThreadManage(){
void CThreadManage::SetParallelNum(int num){
void CThreadManage::Run(CJob* job,void* jobdata){
void CThreadManage::TerminateAll(void){
CThread
CThread 类实现了对Linux中线程操作的封装,它是所有线程的基类,也是一个抽象类,提供了一个抽象接口Run,所有的CThread都必须实现该Run方法。CThread的定义如下所示:
Semaphore m_ThreadSemaphore; //the inner semaphore, which is used to realize
bool m_Detach; //The thread is detached
bool m_CreateSuspended; //if suspend after creating
ThreadState m_ThreadState; //the state of the thread
void SetErrcode(int errcode){m_ErrCode = errcode;}
static void* ThreadFunction(void*);
CThread(bool createsuspended,bool detach);
void SetThreadState(ThreadState state){m_ThreadState = state;}
bool Terminate(void); //Terminate the threa
bool Start(void); //Start to execute the thread
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);
线程类的相关操作均十分简单。线程的执行入口是从Start()函数开始,其将调用函数ThreadFunction,ThreadFunction再调用实际的Run函数,执行实际的任务。
CThreadPool
CThreadPool是线程的承载容器,一般可以将其实现为堆栈、单向队列或者双向队列。在我们的系统中我们使用STL Vector对线程进行保存。CThreadPool的实现代码如下:
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;
CWorkerThread* GetIdleThread(void);
void AppendToIdleList(CWorkerThread* jobthread);
void MoveToBusyList(CWorkerThread* idlethread);
void MoveToIdleList(CWorkerThread* busythread);
void DeleteIdleThread(int num);
void CreateIdleThread(int num);
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
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
vector<CWorkerThread*> m_ThreadList;
vector<CWorkerThread*> m_BusyList; //Thread List
vector<CWorkerThread*> m_IdleList; //Idle List
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 Run(CJob* job,void* jobdata);
CWorkerThread* thr = new CWorkerThread();
CThreadPool::CThreadPool(int initnum)
assert(initnum>0 && initnum<=30);
m_AvailLow = initnum-10>0?initnum-10:3;
m_InitNum=m_AvailNum = initnum ;
CWorkerThread* thr = new CWorkerThread();
thr->Start(); //begin the thread,the thread wait for job
void CThreadPool::TerminateAll()
for(int i=0;i < m_ThreadList.size();i++) {
CWorkerThread* thr = m_ThreadList[i];
CWorkerThread* CThreadPool::GetIdleThread(void)
CWorkerThread* thr = (CWorkerThread*)m_IdleList.front();
printf("Get Idle thread %dn",thr->GetThreadID());
//add an idle thread to idle list
void CThreadPool::AppendToIdleList(CWorkerThread* jobthread)
m_IdleList.push_back(jobthread);
m_ThreadList.push_back(jobthread);
//move and idle thread to busy thread
void CThreadPool::MoveToBusyList(CWorkerThread* idlethread)
m_BusyList.push_back(idlethread);
vector<CWorkerThread*>::iterator pos;
pos = find(m_IdleList.begin(),m_IdleList.end(),idlethread);
void CThreadPool::MoveToIdleList(CWorkerThread* busythread)
m_IdleList.push_back(busythread);
vector<CWorkerThread*>::iterator pos;
pos = find(m_BusyList.begin(),m_BusyList.end(),busythread);
//create num idle thread and put them to idlelist
void CThreadPool::CreateIdleThread(int num)
CWorkerThread* thr = new CWorkerThread();
thr->Start(); //begin the thread,the thread wait for job
void CThreadPool::DeleteIdleThread(int num)
printf("Enter into CThreadPool::DeleteIdleThreadn");
printf("Delete Num is %dn",num);
thr = (CWorkerThread*)m_IdleList.front();
printf("Get Idle thread %dn",thr->GetThreadID());
vector<CWorkerThread*>::iterator pos;
pos = find(m_IdleList.begin(),m_IdleList.end(),thr);
printf("The idle thread available num:%d n",m_AvailNum);
printf("The idlelist num:%d n",m_IdleList.size());
void CThreadPool::Run(CJob* job,void* jobdata)
//if the busy thread num adds to m_MaxNum,so we should wait
if(m_IdleList.size()<m_AvailLow)
if(GetAllNum()+m_InitNum-m_IdleList.size() < m_MaxNum )
CreateIdleThread(m_InitNum-m_IdleList.size());
CreateIdleThread(m_MaxNum-GetAllNum());
CWorkerThread* idlethr = GetIdleThread();
printf("Job is set to thread %d n",idlethr->GetThreadID());
CWorkerThread* thr = new CWorkerThread();
thr->Start(); //begin the thread,the thread wait for job
在该代码中,我们将创建m_InitNum个线程,创建之后即调用AppendToIdleList放入Idle链表中,由于目前没有任务分发给这些线程,因此线程执行Start后将自己挂起。
事实上,线程池中容纳的线程数目并不是一成不变的,其会根据执行负载进行自动伸缩。为此在CThreadPool中设定四个变量:
m_MaxNum:当前线程池中所允许并发存在的线程的最大数目。
m_AvailLow:当前线程池中所允许存在的空闲线程的最小数目,如果空闲数目低于该值,表明负载可能过重,此时有必要增加空闲线程池的数目。实现中我们总是将线程调整为m_InitNum个。
m_AvailHigh:当前线程池中所允许的空闲的线程的最大数目,如果空闲数目高于该值,表明当前负载可能较轻,此时将删除多余的空闲线程,删除后调整数也为m_InitNum个。
线程池在接受到新的任务之后,线程池首先要检查是否有足够的空闲池可用。检查分为三个步骤:
(1)检查当前处于忙碌状态的线程是否达到了设定的最大值m_MaxNum,如果达到了,表明目前没有空闲线程可用,而且也不能创建新的线程,因此必须等待直到有线程执行完毕返回到空闲队列中。
if(GetAllNum()+m_InitNum-m_IdleList.size() < m_MaxNum )
CreateIdleThread(m_InitNum-m_IdleList.size());
CreateIdleThread(m_MaxNum-GetAllNum());
如果创建后总数不超过m_MaxNum,则创建后的线程为m_InitNum;如果超过了,则只创建( m_MaxNum-当前线程总数)个。
(3)调用GetIdleThread方法查找空闲线程。如果当前没有空闲线程,则挂起;否则将任务指派给该线程,同时将其移入忙碌队列。
当线程执行完毕后,其会调用MoveToIdleList方法移入空闲链表中,其中还调用m_IdleCond.Signal()方法,唤醒GetIdleThread()中可能阻塞的线程。
CWorkerThread
CWorkerThread是CThread的派生类,是事实上的工作线程。在CThreadPool的构造函数中,我们创建了一定数量的CWorkerThread。一旦这些线程创建完毕,我们将调用Start()启动该线程。Start方法最终会调用Run方法。Run方法是个无限循环的过程。在没有接受到实际的任务的时候,m_Job为NULL,此时线程将调用Wait方法进行等待,从而处于挂起状态。一旦线程池将具体的任务分发给该线程,其将被唤醒,从而通知线程从挂起的地方继续执行。CWorkerThread的完整定义如下:
class CWorkerThread:public CThread
{
private:
CThreadPool* m_ThreadPool;
CJob* m_Job;
void* m_JobData;
CThreadMutex m_VarMutex;
bool m_IsEnd;
protected:
public:
CCondition m_JobCond;
CThreadMutex m_WorkMutex;
CWorkerThread();
virtual ~CWorkerThread();
void Run();
void SetJob(CJob* job,void* jobdata);
CJob* GetJob(void){return m_Job;}
void SetThreadPool(CThreadPool* thrpool);
CThreadPool* GetThreadPool(void){return m_ThreadPool;}
};
CWorkerThread::CWorkerThread()
{
m_Job = NULL;
m_JobData = NULL;
m_ThreadPool = NULL;
m_IsEnd = false;
}
CWorkerThread::~CWorkerThread()
{
if(NULL != m_Job)
delete m_Job;
if(m_ThreadPool != NULL)
delete m_ThreadPool;
}
void CWorkerThread::Run()
{
SetThreadState(THREAD_RUNNING);
for(;;)
{
while(m_Job == NULL)
m_JobCond.Wait();
m_Job->Run(m_JobData);
m_Job->SetWorkThread(NULL);
m_Job = NULL;
m_ThreadPool->MoveToIdleList(this);
if(m_ThreadPool->m_IdleList.size() > m_ThreadPool->GetAvailHighNum())
{
m_ThreadPool->DeleteIdleThread(m_ThreadPool->m_IdleList.size()-m_T
hreadPool->GetInitNum());
}
m_WorkMutex.Unlock();
}
}
void CWorkerThread::SetJob(CJob* job,void* jobdata)
{
m_VarMutex.Lock();
m_Job = job;
m_JobData = jobdata;
job->SetWorkThread(this);
m_VarMutex.Unlock();
m_JobCond.Signal();
}
void CWorkerThread::SetThreadPool(CThreadPool* thrpool)
{
m_VarMutex.Lock();
m_ThreadPool = thrpool;
m_VarMutex.Unlock();
}
当线程执行任务之前首先必须判断空闲线程的数目是否低于m_AvailLow,如果低于,则必须创建足够的空闲线程,使其数目达到m_InitNum个,然后将调用MoveToBusyList()移出空闲队列,移入忙碌队列。当任务执行完毕后,其又调用MoveToIdleList()移出忙碌队列,移入空闲队列,等待新的任务。
除了Run方法之外,CWorkerThread中另外一个重要的方法就是SetJob,该方法将实际的任务赋值给线程。当没有任何执行任务即m_Job为NULL的时候,线程将调用m_JobCond.Wait进行等待。一旦Job被赋值给线程,其将调用m_JobCond.Signal方法唤醒该线程。由于m_JobCond属于线程内部的变量,每个线程都维持一个m_JobCond,只有得到任务的线程才被唤醒,没有得到任务的将继续等待。无论一个线程何时被唤醒,其都将从等待的地方继续执行m_Job->Run(m_JobData),这是线程执行实际任务的地方。
在线程执行给定Job期间,我们必须防止另外一个Job又赋给该线程,因此在赋值之前,通过m_VarMutex进行锁定, Job执行期间,其于的Job将不能关联到该线程;任务执行完毕,我们调用m_VarMutex.Unlock()进行解锁,此时,线程又可以接受新的执行任务。
在线程执行任务结束后返回空闲队列前,我们还需要判断当前空闲队列中的线程是否高于m_AvailHigh个。如果超过m_AvailHigh,则必须从其中删除(m_ThreadPool->m_IdleList.size()-m_ThreadPool->GetInitNum())个线程,使线程数目保持在m_InitNum个。
CJob
CJob类相对简单,其封装了任务的基本的属性和方法,其中最重要的是Run方法,代码如下:
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::CJob(void)
:m_pWorkThread(NULL)
,m_JobNo(0)
,m_JobName(NULL)
{
}
CJob::~CJob(){
if(NULL != m_JobName)
free(m_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);
}
}
线程池使用示例
至此我们给出了一个简单的与具体任务无关的线程池框架。使用该框架非常的简单,我们所需要的做的就是派生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。
线程池使用后记
线程池适合场合
事实上,线程池并不是万能的。它有其特定的使用场合。线程池致力于减少线程本身的开销对应用所产生的影响,这是有前提的,前提就是线程本身开销与线程执行任务相比不可忽略。如果线程本身的开销相对于线程任务执行开销而言是可以忽略不计的,那么此时线程池所带来的好处是不明显的,比如对于FTP服务器以及Telnet服务器,通常传送文件的时间较长,开销较大,那么此时,我们采用线程池未必是理想的方法,我们可以选择“即时创建,即时销毁”的策略。
总之线程池通常适合下面的几个场合:
(1) 单位时间内处理任务频繁而且任务处理时间短
(2) 对实时性要求较高。如果接受到任务后在创建线程,可能满足不了实时要求,因此必须采用线程池进行预创建。
(3) 必须经常面对高突发性事件,比如Web服务器,如果有足球转播,则服务器将产生巨大的冲击。此时如果采取传统方法,则必须不停的大量产生线程,销毁线程。此时采用动态线程池可以避免这种情况的发生。
结束语
本文给出了一个简单的通用的与任务无关的线程池的实现,通过该线程池能够极大的简化Linux下多线程的开发工作。该线程池的进一步完善开发工作还在进行中,希望能够得到你的建议和支持。
参考资料
http://www-900.ibm.com/developerWorks/cn/java/j-jtp0730/index.shtml
POSIX多线程程序设计,David R.Butenhof 译者:于磊 曾刚,中国电力出版社
C++面向对象多线程编程,CAMERON HUGHES等著 周良忠译,人民邮电出版社
Java Pro,结合线程和分析器池,Edy Yu
关于作者
张中庆,西安交通大学软件所,在读硕士,目前研究方向为分布式网络与移动中间件,对Linux极其爱好,可以通过flydish1234@sina.com.cn与我联系。
线程池原理及创建(C++实现)相关推荐
- C++线程池原理及创建(转)
C++线程池原理及创建(转) 来自http://www.cnblogs.com/cpper-kaixuan/p/3640485.html 本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进 ...
- 线程池原理及创建并C++实现
本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进行了高层次的抽象,使之与具体的执行任务无关.另外该线程池具有动态伸缩性,它能根据执行任务的轻重自动调整线程池中线程的数量.文章的最后,我们 ...
- C++线程池原理及创建
本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进行了高层次的抽象,使之与具体的执行任务无关.另外该线程池具有动态伸缩性,它能根据执行任务的轻重自动调整线程池中线程的数量.文章的最后,我们 ...
- 线程池原理及创建(转)
本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进行了高层次的抽象,使之与具体的执行任务无关.另外该线程池具有动态伸缩性,它能根据执行任务的轻重自动调整线程池中线程的数量.文章的最后,我们 ...
- [线程池][完整实现] 转:线程池原理及创建(C++实现)
在实际任务中经常会用到多线程,但是没用过线程池,我以前的方法比较暴力:1)创建线程,不断切换任务,保持线程一直运行,直到所有任务结束:2)不断的创建和销毁线程:一般都用1方法.最近和别人聊天,谈到线程 ...
- [C++][线程池][完整实现] 转:线程池原理及创建(C++实现)
文章的主要框架是参考这篇文档的,http://jacky-dai.iteye.com/blog/1090285, 关于作者 张中庆,西安交通大学软件所,在读硕士,目前研究方向为分布式网络与移动中间件, ...
- python3 线程池源码解析_5分钟看懂系列:Python 线程池原理及实现
概述 传统多线程方案会使用"即时创建, 即时销毁"的策略.尽管与创建进程相比,创建线程的时间已经大大的缩短,但是如果提交给线程的任务是执行时间较短,而且执行次数极其频繁,那么服务器 ...
- Java多线程闲聊(四):阻塞队列与线程池原理
Java多线程闲聊(四)-阻塞队列与线程池原理 前言 复用永远是人们永恒的主题,这能让我们更好地避免重复制造轮子. 说到多线程,果然还是绕不开线程池,那就来聊聊吧. 人们往往相信,世界是存在一些规律的 ...
- java 线程池原理分析
一.为什么使用线程池 1.降低资源消耗,减少线程创建和销毁次数,每个工作线程可以重复利用,执行多个任务 2.可根据系统承受能力,调整工作线程的数目,防止消耗过多的内存 二.java 线程池使用 Exe ...
最新文章
- 【Vue】Vue1.0+Webpack1+Gulp项目升级构建方案的踩坑路
- uefi启动u盘安装系统_技嘉uefi启动怎么安装win8.1系统【安装教程】
- 纪念逝去的头发--一次debug经历
- wince 6.0 设备管理器架构
- 【转】理解小波消失矩
- ASP.NET Core微服务(五)——【vue脚手架解析接口】
- HarmonyOS之AI能力·图像超分辨率
- Android应用安全开发之浅谈网页打开APP
- 【华为云技术分享】【论文阅读】增量学习近期进展及未来趋势预测
- Java前端顺序,java种初始化顺序
- 百度小程序html解析图片过大_如何快速高效爬取谷歌百度必应的图片
- 用jsp编写一个猜26个小写英文字母的web小游戏
- 汉诺塔实例python_python递归版汉诺塔实例 python汉诺塔递归算法
- 《做难而正确的事》有感
- CS相关书籍的书评[转载]
- 用CSS实现阴阳八卦图等图形
- python开发积累
- 代谢组学通路富集分析
- 从网页端进入1加(one plus)手机云空间
- 大学物理第三版朱峰课后答案详解_大学物理答案(第三版)汇总