Linux下的通用线程池创建(1)_系统/服务器_新闻_云联天下-智慧云谷-智慧家居-物联网-智能家居,智能家居系统,智能家居加盟解决方案-智慧云谷云计算有限公司-智慧家居接口|传感器接口|智能开关接口|工业控制接口

本文给出了一个通用的线程池框架，该框架将与线程执行相关的任务进行了高层次的抽象，使之与具体的执行任务无关。另外该线程池具有动态伸缩性，它能根据执行任务的轻重自动调整线程池中线程的数量。文章的最后，我们给出一个简单示例程序，通过该示例程序，我们会发现，通过该线程池框架执行多线程任务是多么的简单。

为什么需要线程池

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

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

我们将传统方案中的线程执行过程分为三个过程：T1、T2、T3。

T1：线程创建时间

T2：线程执行时间，包括线程的同步等时间

T3：线程销毁时间

那么我们可以看出，线程本身的开销所占的比例为(T1+T3) / (T1+T2+T3)。如果线程执行的时间很短的话，这比开销可能占到20%-50%左右。如果任务执行时间很频繁的话，这笔开销将是不可忽略的。

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

因此线程池的出现正是着眼于减少线程池本身带来的开销。线程池采用预创建的技术，在应用程序启动之后，将立即创建一定数量的线程(N1)，放入空闲队列中。这些线程都是处于阻塞（Suspended）状态，不消耗CPU，但占用较小的内存空间。当任务到来后，缓冲池选择一个空闲线程，把任务传入此线程中运行。当N1个线程都在处理任务后，缓冲池自动创建一定数量的新线程，用于处理更多的任务。在任务执行完毕后线程也不退出，而是继续保持在池中等待下一次的任务。当系统比较空闲时，大部分线程都一直处于暂停状态，线程池自动销毁一部分线程，回收系统资源。

基于这种预创建技术，线程池将线程创建和销毁本身所带来的开销分摊到了各个具体的任务上，执行次数越多，每个任务所分担到的线程本身开销则越小，不过我们另外可能需要考虑进去线程之间同步所带来的开销。

构建线程池框架

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

线程池管理器:用于创建并管理线程池

工作线程: 线程池中实际执行的线程

任务接口: 尽管线程池大多数情况下是用来支持网络服务器，但是我们将线程执行的任务抽象出来，形成任务接口，从而是的线程池与具体的任务无关。

任务队列:线程池的概念具体到实现则可能是队列，链表之类的数据结构，其中保存执行线程。

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

CJob是所有的任务的基类，其提供一个接口Run，所有的任务类都必须从该类继承，同时实现Run方法。该方法中实现具体的任务逻辑。

CThread是Linux中线程的包装，其封装了Linux线程最经常使用的属性和方法，它也是一个抽象类，是所有线程类的基类，具有一个接口Run。

CWorkerThread是实际被调度和执行的线程类，其从CThread继承而来，实现了CThread中的Run方法。

CThreadPool是线程池类，其负责保存线程，释放线程以及调度线程。

CThreadManage是线程池与用户的直接接口，其屏蔽了内部的具体实现。

CThreadMutex用于线程之间的互斥。

CCondition则是条件变量的封装，用于线程之间的同步。

它们的类的继承关系如下图所示：

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

理解系统组件

下面我们分开来了解系统中的各个组件。

CThreadManage

CThreadManage的功能非常简单，其提供最简单的方法，其类定义如下：

class CThreadManage

{

private:

CThreadPool* m_Pool;

int m_NumOfThread;

protected:

public:

void SetParallelNum(int num);

CThreadManage();

CThreadManage(int num);

virtual ~CThreadManage();

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

ce="Courier New" size=2>}

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

};

线程的状态可以分为四种，空闲、忙碌、挂起、终止(包括正常退出和非正常退出)。由于目前Linux线程库不支持挂起操作，因此，我们的此处的挂起操作类似于暂停。如果线程创建后不想立即执行任务，那么我们可以将其“暂停”，如果需要运行，则唤醒。有一点必须注意的是，一旦线程开始执行任务，将不能被挂起，其将一直执行任务至完毕。

线程类的相关操作均十分简单。线程的执行入口是从Start()函数开始，其将调用函数ThreadFunction，ThreadFunction再调用实际的Run函数，执行实际的任务。

CThreadPool

CThreadPool是线程的承载容器，一般可以将其实现为堆栈、单向队列或者双向队列。在我们的系统中我们使用STL Vector对线程进行保存。CThreadPool的实现代码如下：

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

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

};

CThreadPool::CThreadPool()

{

m_MaxNum = 50;

m_AvailLow = 5;

m_InitNum=m_AvailNum = 10 ;

m_AvailHigh = 20;

m_BusyList.clear();

m_IdleList.clear();

for(int i=0;i<m_InitNum;i++){

CWorkerThread* thr = new CWorkerThread();

thr->SetThreadPool(this);

AppendToIdleList(thr);

thr->Start();

}

CThreadPool::CThreadPool(int initnum)

{

assert(initnum>0 && initnum<=30);

m_MaxNum = 30;

m_AvailLow = initnum-10>0?initnum-10:3;

m_InitNum=m_AvailNum = initnum ;

m_AvailHigh = initnum+10;

m_BusyList.clear();

m_IdleList.clear();

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

}

CThreadPool::~CThreadPool()

{

TerminateAll();

}

void CThreadPool::TerminateAll()

{

for(int i=0;i < m_ThreadList.size();i++) {

CWorkerThread* thr = m_ThreadList[i];

thr->Join();

}

return;

}

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 %dn",thr->GetThreadID());

m_IdleMutex.Unlock();

return thr;

}

m_IdleMutex.Unlock();

return NULL;

}

//add an idle thread to idle list

void CThreadPool::AppendToIdleList(CWorkerThread* jobthread)

{

m_IdleMutex.Lock();

push_back(jobthread);

m_ThreadList.push_back(jobthread);

m_IdleMutex.Unlock();

}

//move and idle thread to busy thread

void CThreadPool::MoveToBusyList(CWorkerThread* idlethread)

{

m_BusyMutex.Lock();

m_BusyList.push_back(idlethread);

m_AvailNum--;

m_BusyMutex.Unlock();

m_IdleMutex.Lock();

vector<CWorkerThread*>::iterator pos;

pos = find(m_IdleList.begin(),m_IdleList.end(),idlethread);

if(pos !=m_IdleList.end())

m_IdleList.erase(pos);

m_IdleMutex.Unlock();

}

void CThreadPool::MoveToIdleList(CWorkerThread* busythread)

{

m_IdleMutex.Lock();

m_IdleList.push_back(busythread);

m_AvailNum++;

m_IdleMutex.Unlock();

m_BusyMutex.Lock();

vector<CWorkerThread*>::iterator pos;

pos = find(m_BusyList.begin(),m_BusyList.end(),busythread);

if(pos!=m_BusyList.end())

m_BusyList.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)

{

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::DeleteIdleThread(int num)

{

printf("Enter into CThreadPool::DeleteIdleThreadn");

m_IdleMutex.Lock();

printf("Delete Num is %dn",num);

for(int i=0;i<num;i++){

CWorkerThread* thr;

if(m_IdleList.size() > 0 ){

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

if(pos!=m_IdleList.end())

m_IdleList.erase(pos);

m_AvailNum--;

printf("The idle thread available num:%d n",m_AvailNum);

printf("The idlelist num:%d n",m_IdleList.size());

}

m_IdleMutex.Unlock();

}

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

}