本文給出了一個通用的執行緒池框架，該框架將與執行緒執行相關的任務進行了高層次的抽象，使之與具體的執行任務無關。另外該執行緒池具有動態伸縮性，它能根據執行 任務的輕重自動調整執行緒池中執行緒的數量。文章的最後，我們給出一個簡單示例程式，通過該示例程式，我們會發現，通過該執行緒池框架執行多執行緒任務是多麼的簡 單。

為什麼需要執行緒池

        目前的大多數網路伺服器，包括Web伺服器、Email伺服器以及資料庫伺服器等都具有一個共同點，就是單位時間內必須處理數目巨大的連線請求，但處理時間卻相對較短。

傳 統多執行緒方案中我們採用的伺服器模型則是一旦接受到請求之後，即建立一個新的執行緒，由該執行緒執行任務。任務執行完畢後，執行緒退出，這就是是“即時建立，即 時銷燬”的策略。儘管與建立程序相比，建立執行緒的時間已經大大的縮短，但是如果提交給執行緒的任務是執行時間較短，而且執行次數極其頻繁，那麼伺服器將處於 不停的建立執行緒，銷燬執行緒的狀態。

我們將傳統方案中的執行緒執行過程分為三個過程：T1、T2、T3。

T1：執行緒建立時間

T2：執行緒執行時間，包括執行緒的同步等時間

T3：執行緒銷燬時間

        那麼我們可以看出，執行緒本身的開銷所佔的比例為(T1+T3) / (T1+T2+T3)。如果執行緒執行的時間很短的話，這比開銷可能佔到20%-50%左右。如果任務執行時間很頻繁的話，這筆開銷將是不可忽略的。

除此之外，執行緒池能夠減少建立的執行緒個數。通常執行緒池所允許的併發執行緒是有上界的，如果同時需要併發的執行緒數超過上界，那麼一部分執行緒將會等待。而傳統方案中，如果同時請求數目為2000，那麼最壞情況下，系統可能需要產生2000個執行緒。儘管這不是一個很大的數目，但是也有部分機器可能達不到這種要求。

因此執行緒池的出現正是著眼於減少執行緒池本身帶來的開銷。執行緒池採用預建立的技術，在應用程式啟動之後，將立即建立一定數量的執行緒(N1)，放入空閒佇列中。這些執行緒都是處於阻塞（Suspended）狀態，不消耗CPU，但佔用較小的記憶體空間。當任務到來後，緩衝池選擇一個空閒執行緒，把任務傳入此執行緒中執行。當N1個執行緒都在處理任務後，緩衝池自動建立一定數量的新執行緒，用於處理更多的任務。在任務執行完畢後執行緒也不退出，而是繼續保持在池中等待下一次的任務。當系統比較空閒時，大部分執行緒都一直處於暫停狀態，執行緒池自動銷燬一部分執行緒，回收系統資源。

基於這種預建立技術，執行緒池將執行緒建立和銷燬本身所帶來的開銷分攤到了各個具體的任務上，執行次數越多，每個任務所分擔到的執行緒本身開銷則越小，不過我們另外可能需要考慮進去執行緒之間同步所帶來的開銷。

構建執行緒池框架

一般執行緒池都必須具備下面幾個組成部分：

1、執行緒池管理器:用於建立並管理執行緒池

2、工作執行緒: 執行緒池中實際執行的執行緒

3、任務介面: 儘管執行緒池大多數情況下是用來支援網路伺服器，但是我們將執行緒執行的任務抽象出來，形成任務介面，從而是的執行緒池與具體的任務無關。

4、任務佇列:執行緒池的概念具體到實現則可能是佇列，連結串列之類的資料結構，其中儲存執行執行緒。

我們實現的通用執行緒池框架由五個重要部分組成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:
    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);
};

</pre><p></p><p><span style="color:#666666"><span style="white-space:pre"></span></span><span style="color:rgb(102,102,102); white-space:pre"></span><span style="color:#666666">        執行緒的狀態可以分為四種，</span><strong><span style="color:#ff0000">空閒、忙碌、掛起、終止</span></strong><span style="color:#666666">(包括正常退出和非正常退出)。由於目前Linux執行緒庫不支援掛起操作，因此，我們的此處的掛起操作類似於暫停。如果執行緒建立後不想立即執行任務，那麼我們可以將其“暫停”，如果需要執行，則喚醒。有一點必須注意的是，一旦執行緒開始執行任務，將不能被掛起，其將一直執行任務至完畢。</span></p><p style="color:rgb(102,102,102)"><span style="white-space:pre"></span><span style="white-space:pre"></span>執行緒類的相關操作均十分簡單。執行緒的執行入口是從Start()函式開始，其將呼叫函式ThreadFunction，ThreadFunction再呼叫實際的Run函式，執行實際的任務。</p><p style="color:rgb(102,102,102)"> </p><p style="color:rgb(102,102,102)">CThreadPool</p><p><span style="color:#666666"><span style="white-space:pre"></span></span><span style="color:rgb(102,102,102); white-space:pre"></span><span style="color:#666666">        CThreadPool是執行緒的承載容器，一般可以將其實現為</span><strong><span style="color:#ff6666">堆疊、單向佇列或者雙向佇列</span></strong><span style="color:#666666">。在我們的系統中我們使用STL Vector對執行緒進行儲存。CThreadPool的實現程式碼如下：</span></p><p style="color:rgb(102,102,102)"> </p><p style="color:rgb(102,102,102)"></p><pre name="code" class="cpp">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);
};
 
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 %d/n",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();
    m_IdleList.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::DeleteIdleThread/n");
    m_IdleMutex.Lock();
    printf("Delete Num is %d/n",num);
    for(int i=0;i<num;i++)
    {
        CWorkerThread* thr;
        if(m_IdleList.size() > 0 )
        {
            thr = (CWorkerThread*)m_IdleList.front();
            printf("Get Idle thread %d/n",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);
    }
}

        在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個執行緒，建立之後即呼叫AppendToIdleList放入Idle連結串列中，由於目前沒有任務分發給這些執行緒，因此執行緒執行Start後將自己掛起。

事實上，執行緒池中容納的執行緒數目並不是一成不變的，其會根據執行負載進行自動伸縮。為此在CThreadPool中設定四個變數：

m_InitNum：處世建立時執行緒池中的執行緒的個數。

m_MaxNum:當前執行緒池中所允許併發存在的執行緒的最大數目。

m_AvailLow:當前執行緒池中所允許存在的空閒執行緒的最小數目，如果空閒數目低於該值，表明負載可能過重，此時有必要增加空閒執行緒池的數目。實現中我們總是將執行緒調整為m_InitNum個。

m_AvailHigh：當前執行緒池中所允許的空閒的執行緒的最大數目，如果空閒數目高於該值，表明當前負載可能較輕，此時將刪除多餘的空閒執行緒，刪除後調整數也為m_InitNum個。

m_AvailNum：目前執行緒池中實際存在的執行緒的個數，其值介於m_AvailHigh和m_AvailLow之間。如果執行緒的個數始終維持在m_AvailLow和m_AvailHigh之間，則執行緒既不需要建立，也不需要刪除，保持平衡狀態。因此如何設定m_AvailLow和m_AvailHigh的值，使得執行緒池最大可能的保持平衡態，是執行緒池設計必須考慮的問題。

執行緒池在接受到新的任務之後，執行緒池首先要檢查是否有足夠的空閒池可用。檢查分為三個步驟：

(1)檢查當前處於忙碌狀態的執行緒是否達到了設定的最大值m_MaxNum，如果達到了，表明目前沒有空閒執行緒可用，而且也不能建立新的執行緒，因此必須等待直到有執行緒執行完畢返回到空閒佇列中。

(2)如果當前的空閒執行緒數目小於我們設定的最小的空閒數目m_AvailLow，則我們必須建立新的執行緒，預設情況下，建立後的執行緒數目應該為m_InitNum，因此建立的執行緒數目應該為( 當前空閒執行緒數與m_InitNum);但是有一種特殊情況必須考慮，就是現有的執行緒總數加上建立後的執行緒數可能超過m_MaxNum，因此我們必須對執行緒的建立區別對待

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

如果建立後總數不超過m_MaxNum，則建立後的執行緒為m_InitNum；如果超過了，則只建立( m_MaxNum-當前執行緒總數 )個。

(3)呼叫GetIdleThread方法查詢空閒執行緒。如果當前沒有空閒執行緒，則掛起；否則將任務指派給該執行緒，同時將其移入忙碌佇列。

    當執行緒執行完畢後，其會呼叫MoveToIdleList方法移入空閒連結串列中，其中還呼叫m_IdleCond.Signal()方法，喚醒GetIdleThread()中可能阻塞的執行緒。

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類，將需要完成的任務實現在Run方法中。然後將該Job交由CThreadManage去執行。下面我們給出一個簡單的示例程式</div>

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

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伺服器，如果有足球轉播，則伺服器將產生巨大的衝擊。此時如果採取傳統方法，則必須不停的大量產生執行緒，銷燬執行緒。此時採用動態執行緒池可以避免這種情況的發生。

博主原來的格式被我修改了下，準備測試下里面程式是否正確，測完在更新下體悟。