1、LINUX c+線程池框架本文給出了一個通用的線程池框架，該框架將與線程執(zhí)行相關(guān)的任務(wù)進(jìn)行了高層次的抽象，使之與具體的執(zhí)行任務(wù)無關(guān)。另外該線程池具有動態(tài)伸縮性，它能根據(jù)執(zhí)行任務(wù)的輕重自動調(diào)整線程池中線程的數(shù)量。文章的最后，我們給出一個簡單示例程序，通過該示例程序，我們會發(fā)現(xiàn)，通過該線程池框架執(zhí)行多線程任務(wù)是多么的簡單。為什么需要線程池目前的大多數(shù)網(wǎng)絡(luò)服務(wù)器，包括Web服務(wù)器、Email服務(wù)器以及數(shù)據(jù)庫服務(wù)器等都具有一個共同點(diǎn)，就是單位時間內(nèi)必須處理數(shù)目巨大的連接請求，但處理時間卻相對較短。傳統(tǒng)多線程方案中我們采用的服務(wù)器模型則是一旦接受到請求之后，即創(chuàng)建一個新的線程，由該線程執(zhí)行任務(wù)。任務(wù)執(zhí)行

2、完畢后，線程退出，這就是是“即時創(chuàng)建，即時銷毀”的策略。盡管與創(chuàng)建進(jìn)程相比，創(chuàng)建線程的時間已經(jīng)大大的縮短，但是如果提交給線程的任務(wù)是執(zhí)行時間較短，而且執(zhí)行次數(shù)極其頻繁，那么服務(wù)器將處于不停的創(chuàng)建線程，銷毀線程的狀態(tài)。我們將傳統(tǒng)方案中的線程執(zhí)行過程分為三個過程：T1、T2、T3。T1：線程創(chuàng)建時間 T2：線程執(zhí)行時間，包括線程的同步等時間T3：線程銷毀時間那么我們可以看出，線程本身的開銷所占的比例為(T1+T3) / (T1+T2+T3)。如果線程執(zhí)行的時間很短的話，這比開銷可能占到20%-50%左右。如果任務(wù)執(zhí)行時間很頻繁的話，這筆開銷將是不可忽略的。 除此之外，線程池能夠減少創(chuàng)建的線程個數(shù)。

3、通常線程池所允許的并發(fā)線程是有上界的，如果同時需要并發(fā)的線程數(shù)超過上界，那么一部分線程將會等待。而傳統(tǒng)方案中，如果同時請求數(shù)目為2000，那么最壞情況下，系統(tǒng)可能需要產(chǎn)生2000個線程。盡管這不是一個很大的數(shù)目，但是也有部分機(jī)器可能達(dá)不到這種要求。因此線程池的出現(xiàn)正是著眼于減少線程池本身帶來的開銷。線程池采用預(yù)創(chuàng)建的技術(shù)，在應(yīng)用程序啟動之后，將立即創(chuàng)建一定數(shù)量的線程(N1)，放入空閑隊(duì)列中。這些線程都是處于阻塞（Suspended）狀態(tài)，不消耗CPU，但占用較小的內(nèi)存空間。當(dāng)任務(wù)到來后，緩沖池選擇一個空閑線程，把任務(wù)傳入此線程中運(yùn)行。當(dāng)N1個線程都在處理任務(wù)后，緩沖池自動創(chuàng)建一定數(shù)量的新線程，

4、用于處理更多的任務(wù)。在任務(wù)執(zhí)行完畢后線程也不退出，而是繼續(xù)保持在池中等待下一次的任務(wù)。當(dāng)系統(tǒng)比較空閑時，大部分線程都一直處于暫停狀態(tài)，線程池自動銷毀一部分線程，回收系統(tǒng)資源?；谶@種預(yù)創(chuàng)建技術(shù)，線程池將線程創(chuàng)建和銷毀本身所帶來的開銷分?jǐn)偟搅烁鱾€具體的任務(wù)上，執(zhí)行次數(shù)越多，每個任務(wù)所分擔(dān)到的線程本身開銷則越小，不過我們另外可能需要考慮進(jìn)去線程之間同步所帶來的開銷。 構(gòu)建線程池框架一般線程池都必須具備下面幾個組成部分：線程池管理器:用于創(chuàng)建并管理線程池工作線程: 線程池中實(shí)際執(zhí)行的線程任務(wù)接口: 盡管線程池大多數(shù)情況下是用來支持網(wǎng)絡(luò)服務(wù)器，但是我們將線程執(zhí)行的任務(wù)抽象出來，形成任務(wù)接口，從而是的線

5、程池與具體的任務(wù)無關(guān)。任務(wù)隊(duì)列:線程池的概念具體到實(shí)現(xiàn)則可能是隊(duì)列，鏈表之類的數(shù)據(jù)結(jié)構(gòu)，其中保存執(zhí)行線程。 我們實(shí)現(xiàn)的通用線程池框架由五個重要部分組成CThreadManage，CThreadPool，CThread，CJob，CWorkerThread，除此之外框架中還包括線程同步使用的類CThreadMutex和CCondition。CJob是所有的任務(wù)的基類，其提供一個接口Run，所有的任務(wù)類都必須從該類繼承，同時實(shí)現(xiàn)Run方法。該方法中實(shí)現(xiàn)具體的任務(wù)邏輯。CThread是Linux中線程的包裝，其封裝了Linux線程最經(jīng)常使用的屬性和方法，它也是一個抽象類，是所有線程類的基類，具有一個

6、接口Run。CWorkerThread是實(shí)際被調(diào)度和執(zhí)行的線程類，其從CThread繼承而來，實(shí)現(xiàn)了CThread中的Run方法。CThreadPool是線程池類，其負(fù)責(zé)保存線程，釋放線程以及調(diào)度線程。CThreadManage是線程池與用戶的直接接口，其屏蔽了內(nèi)部的具體實(shí)現(xiàn)。CThreadMutex用于線程之間的互斥。CCondition則是條件變量的封裝，用于線程之間的同步。 CThreadManage直接跟客戶端打交道，其接受需要創(chuàng)建的線程初始個數(shù)，并接受客戶端提交的任務(wù)。這兒的任務(wù)是具體的非抽象的任務(wù)。CThreadManage的內(nèi)部實(shí)際上調(diào)用的都是CThreadPool的相關(guān)操作。C

7、ThreadPool創(chuàng)建具體的線程，并把客戶端提交的任務(wù)分發(fā)給CWorkerThread，CWorkerThread實(shí)際執(zhí)行具體的任務(wù)。 理解系統(tǒng)組件下面我們分開來了解系統(tǒng)中的各個組件。 CThreadManageCThreadManage的功能非常簡單，其提供最簡單的方法，其類定義如下：class CThreadManageprivate: CThreadPool*m_Pool; int m_NumOfThread;protected:public: CThreadManage(); CThreadManage(int num); virtual CThreadManage(); void

8、SetParallelNum(int num); void Run(CJob* job,void* jobdata); void TerminateAll(void);其中m_Pool指向?qū)嶋H的線程池；m_NumOfThread是初始創(chuàng)建時候允許創(chuàng)建的并發(fā)的線程個數(shù)。另外Run和TerminateAll方法也非常簡單，只是簡單的調(diào)用CThreadPool的一些相關(guān)方法而已。其具體的實(shí)現(xiàn)如下:CThreadManage:CThreadManage() m_NumOfThread = 10; m_Pool = new CThreadPool(m_NumOfThread);CThreadManage

9、: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 CThreadManag

10、e:TerminateAll(void) m_Pool-TerminateAll();CThreadCThread 類實(shí)現(xiàn)了對Linux中線程操作的封裝，它是所有線程的基類，也是一個抽象類，提供了一個抽象接口Run，所有的CThread都必須實(shí)現(xiàn)該Run方法。CThread的定義如下所示：class CThreadprivate: int m_ErrCode; Semaphore m_ThreadSemaphore; /the inner semaphore, which is used to realize unsigned long m_ThreadID; bool m_Detach; /

11、The thread is detached bool m_CreateSuspended; /if suspend after creating char* m_ThreadName; ThreadState m_ThreadState; /the state of the threadprotected: void SetErrcode(int errcode)m_ErrCode = errcode; static void* ThreadFunction(void*);public: CThread(); CThread(bool createsuspended,bool detach)

12、; 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 GetLas

13、tError(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 Detac

14、h(void); bool Join(void); bool Yield(void); int Self(void);線程的狀態(tài)可以分為四種，空閑、忙碌、掛起、終止(包括正常退出和非正常退出)。由于目前Linux線程庫不支持掛起操作，因此，我們的此處的掛起操作類似于暫停。如果線程創(chuàng)建后不想立即執(zhí)行任務(wù)，那么我們可以將其“暫?！?，如果需要運(yùn)行，則喚醒。有一點(diǎn)必須注意的是，一旦線程開始執(zhí)行任務(wù)，將不能被掛起，其將一直執(zhí)行任務(wù)至完畢。線程類的相關(guān)操作均十分簡單。線程的執(zhí)行入口是從Start()函數(shù)開始，其將調(diào)用函數(shù)ThreadFunction，ThreadFunction再調(diào)用實(shí)際的Run函數(shù)，執(zhí)行

15、實(shí)際的任務(wù)。CThreadPoolCThreadPool是線程的承載容器，一般可以將其實(shí)現(xiàn)為堆棧、單向隊(duì)列或者雙向隊(duì)列。在我們的系統(tǒng)中我們使用STL Vector對線程進(jìn)行保存。CThreadPool的實(shí)現(xiàn)代碼如下：CThreadPool.h#ifndef CTHREADPOOL_H_INCLUDED#define CTHREADPOOL_H_INCLUDEDclass CThreadPoolfriend class CWorkerThread;public: CThreadPool(); CThreadPool(int initnum); virtual CThreadPool(); voi

16、d 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

17、)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);protected: CWorkerThread* GetIdleThread(v

18、oid); 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 un

19、lock 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 u

20、sed to sync idle thread list CCondition m_IdleJobCond; /m_JobCond is used to sync job list CCondition m_MaxNumCond; vector m_ThreadList; vector m_BusyList; /Thread List vector m_IdleList; /Idle Listprivate: unsigned int m_InitNum; /Normal thread num; unsigned int m_MaxNum; /the max thread num that c

21、an 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;#endif / CTHREADPOOL_H_INCLUDEDCThreadPool.cpp#include CThread

22、Pool.hCThreadPool:CThreadPool() m_MaxNum = 50; m_AvailLow = 5; m_AvailHigh = 20; m_InitNum = m_AvailNum = 10 ; m_BusyList.clear(); m_IdleList.clear(); for(int i=0;iSetThreadPool(this); AppendToIdleList(thr); thr-Start(); CThreadPool:CThreadPool(int initnum) assert(initnum0 & initnum0?initnum-10:3; m

23、_InitNum=m_AvailNum = initnum ; m_AvailHigh = initnum+10; m_BusyList.clear(); m_IdleList.clear(); for(int i=0;iSetThreadPool(this); thr-Start(); /begin the thread,the thread wait for job CThreadPool:CThreadPool() TerminateAll();void CThreadPool:TerminateAll() for(int i=0;i Join(); return;CWorkerThre

24、ad* 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;

25、/add an idle thread to idle listvoid 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 threadvoid CThreadPool:MoveToBusyList(CWorkerThread* idlethread) m_Bu

26、syMutex.Lock(); m_BusyList.push_back(idlethread); m_AvailNum-; m_BusyMutex.Unlock(); m_IdleMutex.Lock(); vector: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(CWorkerThrea

27、d* busythread) m_IdleMutex.Lock(); m_IdleList.push_back(busythread); m_AvailNum+; m_IdleMutex.Unlock(); m_BusyMutex.Lock(); vector: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_Ma

28、xNumCond.Signal();/create num idle thread and put them to idlelistvoid CThreadPool:CreateIdleThread(int num) for(int i=0;iSetThreadPool(this); AppendToIdleList(thr); m_VarMutex.Lock(); m_AvailNum+; m_VarMutex.Unlock(); thr-Start(); /begin the thread,the thread wait for job /create num idle thread an

29、d put them to idlelistvoid CThreadPool:CreateIdleThread(int num) for(int i=0;iSetThreadPool(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 CThreadPo

30、ol:DeleteIdleThreadn); m_IdleMutex.Lock(); printf(Delete Num is %dn,num); for(int i=0;i 0 ) thr = (CWorkerThread*)m_IdleList.front(); printf(Get Idle thread %dn,thr-GetThreadID(); vector:iterator pos; pos = find(m_IdleList.begin(),m_IdleList.end(),thr); if(pos!=m_IdleList.end() m_IdleList.erase(pos)

31、; 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_MaxNumC

32、ond.Wait(); if(m_IdleList.size()m_AvailLow) if(GetAllNum()+m_InitNum-m_IdleList.size() 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中存在兩個鏈表，一個是空閑鏈

33、表，一個是忙碌鏈表。Idle鏈表中存放所有的空閑進(jìn)程，當(dāng)線程執(zhí)行任務(wù)時候，其狀態(tài)變?yōu)槊β禒顟B(tài)，同時從空閑鏈表中刪除，并移至忙碌鏈表中。在CThreadPool的構(gòu)造函數(shù)中，我們將執(zhí)行下面的代碼:for(int i=0;iSetThreadPool(this); thr-Start(); /begin the thread,the thread wait for job 在該代碼中，我們將創(chuàng)建m_InitNum個線程，創(chuàng)建之后即調(diào)用AppendToIdleList放入Idle鏈表中，由于目前沒有任務(wù)分發(fā)給這些線程，因此線程執(zhí)行Start后將自己掛起。事實(shí)上，線程池中容納的線程數(shù)目并不是一成不變的

34、，其會根據(jù)執(zhí)行負(fù)載進(jìn)行自動伸縮。為此在CThreadPool中設(shè)定四個變量：m_InitNum：處世創(chuàng)建時線程池中的線程的個數(shù)。m_MaxNum:當(dāng)前線程池中所允許并發(fā)存在的線程的最大數(shù)目。m_AvailLow:當(dāng)前線程池中所允許存在的空閑線程的最小數(shù)目，如果空閑數(shù)目低于該值，表明負(fù)載可能過重，此時有必要增加空閑線程池的數(shù)目。實(shí)現(xiàn)中我們總是將線程調(diào)整為m_InitNum個。m_AvailHigh：當(dāng)前線程池中所允許的空閑的線程的最大數(shù)目，如果空閑數(shù)目高于該值，表明當(dāng)前負(fù)載可能較輕，此時將刪除多余的空閑線程，刪除后調(diào)整數(shù)也為m_InitNum個。m_AvailNum：目前線程池中實(shí)際存在的線程的

35、個數(shù)，其值介于m_AvailHigh和m_AvailLow之間。如果線程的個數(shù)始終維持在m_AvailLow和m_AvailHigh之間，則線程既不需要創(chuàng)建，也不需要刪除，保持平衡狀態(tài)。因此如何設(shè)定m_AvailLow和m_AvailHigh的值，使得線程池最大可能的保持平衡態(tài)，是線程池設(shè)計(jì)必須考慮的問題。線程池在接受到新的任務(wù)之后，線程池首先要檢查是否有足夠的空閑池可用。檢查分為三個步驟： (1)檢查當(dāng)前處于忙碌狀態(tài)的線程是否達(dá)到了設(shè)定的最大值m_MaxNum，如果達(dá)到了，表明目前沒有空閑線程可用，而且也不能創(chuàng)建新的線程，因此必須等待直到有線程執(zhí)行完畢返回到空閑隊(duì)列中。 (2)如果當(dāng)前的空閑

36、線程數(shù)目小于我們設(shè)定的最小的空閑數(shù)目m_AvailLow，則我們必須創(chuàng)建新的線程，默認(rèn)情況下，創(chuàng)建后的線程數(shù)目應(yīng)該為m_InitNum，因此創(chuàng)建的線程數(shù)目應(yīng)該為( 當(dāng)前空閑線程數(shù)與m_InitNum);但是有一種特殊情況必須考慮，就是現(xiàn)有的線程總數(shù)加上創(chuàng)建后的線程數(shù)可能超過m_MaxNum，因此我們必須對線程的創(chuàng)建區(qū)別對待。 if(GetAllNum()+m_InitNum-m_IdleList.size() Run(m_JobData);m_Job-SetWorkThread(NULL);m_Job = NULL;m_ThreadPool-MoveToIdleList(this);if(m_

37、ThreadPool-m_IdleList.size() m_ThreadPool-GetAvailHighNum()m_ThreadPool-DeleteIdleThread(m_ThreadPool-m_IdleList.size()-m_ThreadPool-GetInitNum();m_WorkMutex.Unlock();void CWorkerThread:SetJob(CJob* job,void* jobdata)m_VarMutex.Lock();m_Job = job;m_JobData = jobdata;job-SetWorkThread(this);m_VarMute

38、x.Unlock();m_JobCond.Signal();void CWorkerThread:SetThreadPool(CThreadPool* thrpool)m_VarMutex.Lock();m_ThreadPool = thrpool;m_VarMutex.Unlock();當(dāng)線程執(zhí)行任務(wù)之前首先必須判斷空閑線程的數(shù)目是否低于m_AvailLow，如果低于，則必須創(chuàng)建足夠的空閑線程，使其數(shù)目達(dá)到m_InitNum個，然后將調(diào)用MoveToBusyList()移出空閑隊(duì)列，移入忙碌隊(duì)列。當(dāng)任務(wù)執(zhí)行完畢后，其又調(diào)用MoveToIdleList()移出忙碌隊(duì)列，移入空閑隊(duì)列，等待新的任

39、務(wù)。除了Run方法之外，CWorkerThread中另外一個重要的方法就是SetJob，該方法將實(shí)際的任務(wù)賦值給線程。當(dāng)沒有任何執(zhí)行任務(wù)即m_Job為NULL的時候，線程將調(diào)用m_JobCond.Wait進(jìn)行等待。一旦Job被賦值給線程，其將調(diào)用m_JobCond.Signal方法喚醒該線程。由于m_JobCond屬于線程內(nèi)部的變量，每個線程都維持一個m_JobCond，只有得到任務(wù)的線程才被喚醒，沒有得到任務(wù)的將繼續(xù)等待。無論一個線程何時被喚醒，其都將從等待的地方繼續(xù)執(zhí)行m_Job-Run(m_JobData)，這是線程執(zhí)行實(shí)際任務(wù)的地方。在線程執(zhí)行給定Job期間，我們必須防止另外一個Job

40、又賦給該線程，因此在賦值之前，通過m_VarMutex進(jìn)行鎖定， Job執(zhí)行期間，其于的Job將不能關(guān)聯(lián)到該線程；任務(wù)執(zhí)行完畢，我們調(diào)用m_VarMutex.Unlock()進(jìn)行解鎖，此時，線程又可以接受新的執(zhí)行任務(wù)。在線程執(zhí)行任務(wù)結(jié)束后返回空閑隊(duì)列前，我們還需要判斷當(dāng)前空閑隊(duì)列中的線程是否高于m_AvailHigh個。如果超過m_AvailHigh，則必須從其中刪除(m_ThreadPool-m_IdleList.size()-m_ThreadPool-GetInitNum()個線程，使線程數(shù)目保持在m_InitNum個。CJobCJob類相對簡單，其封裝了任務(wù)的基本的屬性和方法，其中最重要

41、的是Run方法，代碼如下：class CJobprivate: int m_JobNo; /The num was assigned to the job char* m_JobName; /The job name CThread *m_pWorkThread; /The thread associated with the jobpublic: 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;C