線程池原理及創(chuàng)建(C++實現(xiàn))

1、線程池原理及創(chuàng)建（C+實現(xiàn)）時間:2010 02 2514:40:43來源:網(wǎng)絡作者未知 點擊:2963次 本文給出了一個通用的線程池框架，該框架將與線程執(zhí)行相關的任務進行了高層次的抽象，使之 與具體的執(zhí)行任務無關。另外該線程池具有動態(tài)伸縮性，它能根據(jù)執(zhí)行任務的輕重自動調(diào)整線程 池中線程的數(shù)量。文章的最后，我們給出一個本文給出了一個通用的線程池框架，該框架將與線程執(zhí)行相關的任務進行了高層次的抽象，使之 與具體的執(zhí)行任務無關。另外該線程池具有動態(tài)伸縮性，它能根據(jù)執(zhí)行任務的輕重自動調(diào)整線程 池中線程的數(shù)量。文章的最后，我們給出一個簡單示例程序，通過該示例程序，我們會發(fā)現(xiàn)，通 過該線程池框架執(zhí)行多線

2、程任務是多么的簡單。為什么需要線程池目前的大多數(shù)網(wǎng)絡服務器，包括 Web服務器、Email服務器以及數(shù)據(jù)庫服務器等都具有一個共同 點，就是單位時間內(nèi)必須處理數(shù)目巨大的連接請求，但處理時間卻相對較短。傳統(tǒng)多線程方案中我們采用的服務器模型則是一旦接受到請求之后，即創(chuàng)建一個新的線程，由該 線程執(zhí)行任務。任務執(zhí)行完畢后，線程退出，這就是是“即時創(chuàng)建，即時銷毀”的策略。盡管與 創(chuàng)建進程相比，創(chuàng)建線程的時間已經(jīng)大大的縮短，但是如果提交給線程的任務是執(zhí)行時間較短， 而且執(zhí)行次數(shù)極其頻繁，那么服務器將處于不停的創(chuàng)建線程，銷毀線程的狀態(tài)。我們將傳統(tǒng)方案中的線程執(zhí)行過程分為三個過程：T1、 T2、 T3T1:線程

3、創(chuàng)建時間T2:線程執(zhí)行時間，包括線程的同步等時間T3:線程銷毀時間那么我們可以看出，線程本身的開銷所占的比例為(T1+T3)/ (T1+T2+T3。) 如果線程執(zhí)行的時間很短的話，這比開銷可能占到20% 50%左右。如果任務執(zhí)行時間很頻繁的話，這筆開銷將是不可忽略 的。除此之外，線程池能夠減少創(chuàng)建的線程個數(shù)。通常線程池所允許的并發(fā)線程是有上界的，如果同 時需要并發(fā)的線程數(shù)超過上界，那么一部分線程將會等待。而傳統(tǒng)方案中，如果同時請求數(shù)目為 2000，那么最壞情況下，系統(tǒng)可能需要產(chǎn)生 2000 個線程。盡管這不是一個很大的數(shù)目，但是也 有部分機器可能達不到這種要求。因此線程池的出現(xiàn)正是著眼于減少線

4、程池本身帶來的開銷。線程池采用預創(chuàng)建的技術，在應用程 序啟動之后，將立即創(chuàng)建一定數(shù)量的線程(N1)，放入空閑隊列中。這些線程都是處于阻塞(Suspendec)狀態(tài)，不消耗CPU但占用較小的內(nèi)存空間。當任務到來后，緩沖池選擇一個空閑 線程，把任務傳入此線程中運行。當 N1 個線程都在處理任務后，緩沖池自動創(chuàng)建一定數(shù)量的新 線程，用于處理更多的任務。在任務執(zhí)行完畢后線程也不退出，而是繼續(xù)保持在池中等待下一次 的任務。當系統(tǒng)比較空閑時，大部分線程都一直處于暫停狀態(tài)，線程池自動銷毀一部分線程，回 收系統(tǒng)資源?；谶@種預創(chuàng)建技術，線程池將線程創(chuàng)建和銷毀本身所帶來的開銷分攤到了各個具體的任務上， 執(zhí)行次數(shù)

5、越多，每個任務所分擔到的線程本身開銷則越小，不過我們另外可能需要考慮進去線程 之間同步所帶來的開銷。構(gòu)建線程池框架一般線程池都必須具備下面幾個組成部分：線程池管理器 :用于創(chuàng)建并管理線程池工作線程 : 線程池中實際執(zhí)行的線程任務接口 :盡管線程池大多數(shù)情況下是用來支持網(wǎng)絡服務器，但是我們將線程執(zhí)行的任務抽象出來， 形成任務接口，從而是的線程池與具體的任務無關。任務隊列 :線程池的概念具體到實現(xiàn)則可能是隊列，鏈表之類的數(shù)據(jù)結(jié)構(gòu)，其中保存執(zhí)行線程我們實現(xiàn)的通用線程池框架由五個重要部分組成CThreadManage， CThreadPoo，l CThread， CJob，CWorkerThread,

6、除此之外框架中還包括線程同步使用的類CThreadMutex和CCondition。CJob是所有的任務的基類，其提供一個接口 Run,所有的任務類都必須從該類繼承，同時實現(xiàn) Run方法。該方法中實現(xiàn)具體的任務邏輯。CThread是Linux中線程的包裝，其封裝了 Linux線程最經(jīng)常使用的屬性和方法，它也是一個抽象 類,是所有線程類的基類,具有一個接口Run。CWorkerThread是實際被調(diào)度和執(zhí)行的線程類，其從 CThread繼承而來，實現(xiàn)了 CThread中的Run 方法。CThreadPoo是線程池類，其負責保存線程，釋放線程以及調(diào)度線程。CThreadMa nage是線程池與用戶

7、的直接接口，其屏蔽了內(nèi)部的具體實現(xiàn)。CThreadMutex用于線程之間的互斥CCondition 則是條件變量的封裝,用于線程之間的同步它們的類的繼承關系如下圖所示：線程池的時序很簡單，如下圖所示。CThreadMa nage直接跟客戶端打交道，其接受需要創(chuàng)建的線程初始個數(shù)，并接受客戶端提交的任務。這兒的任務是具體的非抽象的任務。CThreadMa nage的內(nèi)部實際上調(diào)用的都是CThreadPool的相關操作。CThreadPoo創(chuàng)建具體的線程，并把客戶端提交 的任務分發(fā)給CWorkerThread, CWorkerThread實際執(zhí)行具體的任務。理解系統(tǒng)組件下面我們分開來了解系統(tǒng)中的各個

8、組件。CThreadManageCThreadMa nage的功能非常簡單，其提供最簡單的方法，其類定義如下:classCThreadManageprivate:CThreadPool* m_Pool;int m_NumOfThread;protected: public:void SetParallelNum(int num);CThreadManage();CThreadManage(intnum);virtual CThreadManage();void Run(CJob*job,void* jobdata);void TerminateAll(void);其中 m_Pool 指向?qū)嶋H的

9、線程池； m_NumOfThread 是初始創(chuàng)建時候允許創(chuàng)建的并發(fā)的線程個數(shù)。 另外Run和TerminateAII方法也非常簡單，只是簡單的調(diào)用 CThreadPool的一些相關方法而已 其具體的實現(xiàn)如下 :CThreadManage:CThreadManage()m_NumOfThread = 10;m_Pool=new CThreadPool(m_NumOfThread);CThreadManage:CThreadManage(intnum)m_NumOfThread = num;m_Pool = new CThreadPool(m_NumOfThread); CThreadManage

10、:CThreadManage()if(NULL!= m_Pool)delete m_Pool;void CThreadManage:SetParallelNum(intnum) m_NumOfThread = num;void CThreadManage:Run(CJob*job,void* jobdata)m_Pool >Run(jobjobdata);void CThreadManage:TerminateAll(void)m_Pool >TerminateAII();CThreadCThread類實現(xiàn)了對Linux中線程操作的封裝，它是所有線程的基類，也是一個抽象類，提供了一

11、 個抽象接口 Run,所有的CThread都必須實現(xiàn)該Run方法。CThread的定義如下所示： classCThread private:int m_ErrCode;Semaphore m_ThreadSemaphore; /the inner semaphore, which is used to realizeunsigned long m_ThreadID;bool m_Detach; /The thread is detachedbool m_CreateSuspended; /if suspend after creatingchar* m_ThreadName;ThreadSta

12、tem_ThreadState; /the state of the thread protected:void SetErrcode(interrcode)m_ErrCode= errcode;static void* ThreadFunction(void*);public:CThread();CThread(boolcreatesuspended,bool detach);virtual CThread();virtual void Run(void)= 0;void SetThreadState(ThreadStatestate)m_ThreadState = state;bool T

13、erminate(void); /Terminate the threabool Start(void); /Start to execute the threadvoid 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)retur

14、n 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);線程的狀態(tài)可以分為四種，空閑、忙碌、掛起、終止 (包括正常退出和非正常退出 )。由于目前Linux線程庫不支持掛起操作，因此，我們的此處的掛起操作

15、類似于暫停。如果線程創(chuàng)建后不想立 即執(zhí)行任務，那么我們可以將其“暫?！?，如果需要運行，則喚醒。有一點必須注意的是，一旦 線程開始執(zhí)行任務，將不能被掛起，其將一直執(zhí)行任務至完畢。線程類的相關操作均十分簡單。線程的執(zhí)行入口是從 Start()函數(shù)開始，其將調(diào)用函數(shù)ThreadFunction， ThreadFunction 再調(diào)用實際的 Run 函數(shù)，執(zhí)行實際的任務。CThreadPoolCThreadPool是線程的承載容器，一般可以將其實現(xiàn)為堆棧、單向隊列或者雙向隊列。在我們的系 統(tǒng)中我們使用STLVector對線程進行保存。CThreadPool的實現(xiàn)代碼如下：classCThreadPoo

16、lfriend classCWorkerThread;private:unsigned int m_MaxNum; /the maxthread numthat cancreateatthesametimeunsignedintm_AvailLow;/The minnumofidlethread thatshoulekeptunsigned int m_AvailHigh; /The maxnumofidlethread that kept at the sametimeunsigned int m_AvailNum; /the normalthread numofidle num;unsig

17、ned 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:CThread

18、Mutex m_BusyMutex; /when visit busylist,use m_BusyMutex to lock and unlockCThreadMutex m_IdleMutex; /when visit idle list,use m_IdleMutex to lock and unlockCThreadMutex m_JobMutex; /when visit job list,use m_JobMutex to lock and unlockCThreadMutex m_VarMutex;CCondition m_BusyCond;/m_BusyCond is used

19、 to sync busy thread listCCondition m_IdleCond; /m_IdleCond is used to sync idle thread listCCondition m_IdleJobCond; /m_JobCond is used to sync job listCCondition m_MaxNumCond;vector<CWorkerThread*> m_ThreadList;vector<CWorkerThread*>m_BusyList;/Thread Listvector<CWorkerThread*>m_

20、IdleList; /Idle ListCThreadPool();CThreadPool(intinitnum);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

21、 = 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(

22、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

23、>Start();CThreadPool:CThreadPool(intinitnum)assert(initnum>0 && initnum<=30);m_MaxNum = 30;m_AvailLow = i nitnum 10>0?i nitn um 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 = newC

24、WorkerThread();AppendToIdleList(thr);thr >SetThreadPool(this);thr >Start();/begin the thread,the thread wait for jobCThreadPool:CThreadPool()TerminateAll();void CThreadPool:TerminateAll()for(int i=0;i < m_ThreadList.size();i+)CWorkerThread* thr = m_ThreadListi;thr >Join();return;CWorkerT

25、hread* 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 >GetThreadlD();m_IdleMutex.Unlock();return thr;m_IdleMutex.Unlock();ret

26、urn NULL;/add an idle thread to idle listvoid CThreadPool:AppendToIdleList(CWorkerThread*jobthread)m_IdleList.push_back(jobthread);m_IdleMutex.Lock();m_ThreadList.push_back(jobthread);m_IdleMutex.Unlock();/move and idle thread to busy threadvoid CThreadPool:MoveToBusyList(CWorkerThread*idlethread) m

27、_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:MoveToIdl

28、eList(CWorkerThread*busythread)m_IdleList.push_back(busythread);m_IdleMutex.Lock();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

29、();m_IdleCond.Signal();m_MaxNumCond.Signal();/create num idle thread and put them to idlelistvoid CThreadPool:CreateIdleThread(intnum)for(int i=0;i<num;i+)CWorkerThread* thr =new CWorkerThread();thr >SetThreadPool(this);AppendToIdleList(thr);m_VarMutex.Lock();m_AvailNum+;m_VarMutex.Unlock();th

30、r >Start();/begin the thread,the thread wait for jobvoid CThreadPool:DeleteIdleThread(intnum)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 = (CWo

31、rkerThread*)m_IdleList.front();printf("Get Idle thread %dn”,thr >GetThreadlD();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 availablenum:%dn",m_Avail

32、Num);printf("The idlelist num:%d n",m_IdleList.size();m_IdleMutex.Unlock();void CThreadPool:Run(CJob*job,void* jobdata)assert(job!=NULL);/if the busythread num adds to m_MaxNum,so we should waitif(GetBusyNum() = m_MaxNum)m_MaxNumCond.Wait();if(m_IdleList.size()<m_AvailLow)if(GetAIINum()

33、+m_lnitNum m_ldleList.size()v m_MaxNum )CreateldleThread(mnitNum mdIeList.size();elseCreateldleThread(m_MaxNumGetAIINum();CWorkerThread* idlethr = GetldleThread();if(idlethr !=NULL)idlethr >m_WorkMutex.Lock();MoveToBusyList(idlethr);idlethr >SetThreadPool(this);job >SetWorkThread(idlethr);p

34、rintf("Job issetto thread %d n",idlethr >GetThreadlD();idlethr >SetJob(job,jobdata);在CThreadPool中存在兩個鏈表，一個是空閑鏈表，一個是忙碌鏈表。Idle鏈表中存放所有的空閑進程，當線程執(zhí)行任務時候，其狀態(tài)變?yōu)槊β禒顟B(tài)，同時從空閑鏈表中刪除，并移至忙碌鏈表 中。在CThreadPooI的構(gòu)造函數(shù)中，我們將執(zhí)行下面的代碼：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在該代碼中，我們將創(chuàng)建 mnitNum個線程，創(chuàng)建之后即調(diào)用 AppendToldleList放入Idle鏈表 中，由于目前沒有任務分發(fā)給這些線程，因此線程執(zhí)行 Start 后將自己掛起。事實上，線程池中容納的線程數(shù)目并不