??????這兩天在做關(guān)于網(wǎng)絡(luò)蜘蛛的程序,希望可以通過線程池來提高程序的性能,網(wǎng)上搜索了一下,看到這方面的東西還不少,跟大家分享一下!~有許多應(yīng)用程序創(chuàng)建的線程花費了大量時間在睡眠狀態(tài)來等待事件的發(fā)生。還有一些線程進(jìn)入睡眠狀態(tài)后定期被喚醒以輪詢工作方式來改變或者更新狀態(tài)信息。線程池可以讓你更有效地使用線程，它為你的應(yīng)用程序提供一個由系統(tǒng)管理的工作者線程池。至少會有一個線程來監(jiān)聽放到線程池的所有等待操作，當(dāng)?shù)却僮魍瓿珊?，線程池中將會有一個工作者線程來執(zhí)行相應(yīng)的回調(diào)函數(shù)。
? 你也可以把沒有等待操作的工作項目放到線程池中，用QueueUserWorkItem函數(shù)來完成這個工作，把要執(zhí)行的工作項目函數(shù)通過一個參數(shù)傳遞給線程池。工作項目被放到線程池中后，就不能再取消了。
? Timer-queue timers和Registered wait operations也使用線程池來實現(xiàn)。他們的回調(diào)函數(shù)也放在線程池中。你也可以用BindIOCompletionCallback函數(shù)來投遞一個異步IO操作，在IO完成端口上，回調(diào)函數(shù)也是由線程池線程來執(zhí)行。
? 當(dāng)?shù)谝淮握{(diào)用QueueUserWorkItem函數(shù)或者BindIOCompletionCallback函數(shù)的時候，線程池被自動創(chuàng)建，或者Timer-queue timers或者Registered wait operations放入回調(diào)函數(shù)的時候，線程池也可以被創(chuàng)建。線程池可以創(chuàng)建的線程數(shù)量不限，僅受限于可用的內(nèi)存，每一個線程使用默認(rèn)的初始堆棧大小，運行在默認(rèn)的優(yōu)先級上。
? 線程池中有兩種類型的線程：IO線程和非IO線程。IO線程等待在可告警狀態(tài)，工作項目作為APC放到IO線程中。如果你的工作項目需要線程執(zhí)行在可警告狀態(tài)，你應(yīng)該將它放到IO線程。
? 非IO工作者線程等待在IO完成端口上，使用非IO線程比IO線程效率更高，也就是說，只要有可能的話，盡量使用非IO線程。IO線程和非IO線程在異步IO操作沒有完成之前都不會退出。然而，不要在非IO線程中發(fā)出需要很長時間才能完成的異步IO請求。
? 正確使用線程池的方法是，工作項目函數(shù)以及它將會調(diào)用到的所有函數(shù)都必須是線程池安全的。安全的函數(shù)不應(yīng)該假設(shè)線程是一次性線程的或者是永久線程。一般來說，應(yīng)該避免使用線程本地存儲和發(fā)出需要永久線程的異步IO調(diào)用，比如說RegNotifyChangeKeyValue函數(shù)。如果需要在永久線程中執(zhí)行這樣的函數(shù)的話，可以給QueueUserWorkItem傳遞一個選項WT_EXECUTEINPERSISTENTTHREAD。
? 注意，線程池不能兼容COM的單線程套間（STA）模型。
?
? ????為了更深入地講解操作系統(tǒng)實現(xiàn)的線程池的優(yōu)越性，我們首先嘗試著自己實現(xiàn)一個簡單的線程池模型。
?
? ????代碼如下：
??/**//************************************************************************/
??/**//*?Test?Our?own?thread?pool.????????????????????????????????????????????*/
??/**//************************************************************************/
??
??typedef?struct?_THREAD_POOL
??
{
??????HANDLE?QuitEvent;
??????HANDLE?WorkItemSemaphore;
??
??????LONG?WorkItemCount;
??????LIST_ENTRY?WorkItemHeader;
??????CRITICAL_SECTION?WorkItemLock;
??
??????LONG?ThreadNum;
??????HANDLE?*ThreadsArray;
??
??}THREAD_POOL,?*PTHREAD_POOL;
??
??typedef?VOID?(*WORK_ITEM_PROC)(PVOID?Param);
??
??typedef?struct?_WORK_ITEM
??{
??????LIST_ENTRY?List;
??
??????WORK_ITEM_PROC?UserProc;
??????PVOID?UserParam;
??????
??}WORK_ITEM,?*PWORK_ITEM;
??
??
??DWORD?WINAPI?WorkerThread(PVOID?pParam)
??{
??????PTHREAD_POOL?pThreadPool?=?(PTHREAD_POOL)pParam;
??????HANDLE?Events[2];
??????
??????Events[0]?=?pThreadPool->QuitEvent;
??????Events[1]?=?pThreadPool->WorkItemSemaphore;
??
??????for(;;)
??????{
??????????DWORD?dwRet?=?WaitForMultipleObjects(2,?Events,?FALSE,?INFINITE);
??
??????????if(dwRet?==?WAIT_OBJECT_0)
??????????????break;
??
??????????//
??????????//?execute?user's?proc.
??????????//
??
??????????else?if(dwRet?==?WAIT_OBJECT_0?+1)
??????????{
??????????????PWORK_ITEM?pWorkItem;
??????????????PLIST_ENTRY?pList;
??
??????????????EnterCriticalSection(&pThreadPool->WorkItemLock);
??????????????_ASSERT(!IsListEmpty(&pThreadPool->WorkItemHeader));
??????????????pList?=?RemoveHeadList(&pThreadPool->WorkItemHeader);
??????????????LeaveCriticalSection(&pThreadPool->WorkItemLock);
??
??????????????pWorkItem?=?CONTAINING_RECORD(pList,?WORK_ITEM,?List);
??????????????pWorkItem->UserProc(pWorkItem->UserParam);
??
??????????????InterlockedDecrement(&pThreadPool->WorkItemCount);
??????????????free(pWorkItem);
??????????}
??
??????????else
??????????{
??????????????_ASSERT(0);
??????????????break;
??????????}
??????}
??
??????return?0;
??}
??
??BOOL?InitializeThreadPool(PTHREAD_POOL?pThreadPool,?LONG?ThreadNum)
??{
??????pThreadPool->QuitEvent?=?CreateEvent(NULL,?TRUE,?FALSE,?NULL);
??????pThreadPool->WorkItemSemaphore?=?CreateSemaphore(NULL,?0,?0x7FFFFFFF,?NULL);
??????pThreadPool->WorkItemCount?=?0;
??????InitializeListHead(&pThreadPool->WorkItemHeader);
??????InitializeCriticalSection(&pThreadPool->WorkItemLock);
??????pThreadPool->ThreadNum?=?ThreadNum;
??????pThreadPool->ThreadsArray?=?(HANDLE*)malloc(sizeof(HANDLE)?*?ThreadNum);
??
??????for(int?i=0;?i<ThreadNum;?i++)
??????{
??????????pThreadPool->ThreadsArray[i]?=?CreateThread(NULL,?0,?WorkerThread,?pThreadPool,?0,?NULL);
??????}
??
??????return?TRUE;
??}
??
??VOID?DestroyThreadPool(PTHREAD_POOL?pThreadPool)
??{
??????SetEvent(pThreadPool->QuitEvent);
??
??????for(int?i=0;?i<pThreadPool->ThreadNum;?i++)
??????{
??????????WaitForSingleObject(pThreadPool->ThreadsArray[i],?INFINITE);
??????????CloseHandle(pThreadPool->ThreadsArray[i]);
??????}
??
??????free(pThreadPool->ThreadsArray);
??
??????CloseHandle(pThreadPool->QuitEvent);
??????CloseHandle(pThreadPool->WorkItemSemaphore);
??????DeleteCriticalSection(&pThreadPool->WorkItemLock);
??
??????while(!IsListEmpty(&pThreadPool->WorkItemHeader))
??????{
??????????PWORK_ITEM?pWorkItem;
??????????PLIST_ENTRY?pList;
??????????
??????????pList?=?RemoveHeadList(&pThreadPool->WorkItemHeader);
??????????pWorkItem?=?CONTAINING_RECORD(pList,?WORK_ITEM,?List);
??????????
??????????free(pWorkItem);
??????}
??}
??
??BOOL?PostWorkItem(PTHREAD_POOL?pThreadPool,?WORK_ITEM_PROC?UserProc,?PVOID?UserParam)
??{
??????PWORK_ITEM?pWorkItem?=?(PWORK_ITEM)malloc(sizeof(WORK_ITEM));
??????if(pWorkItem?==?NULL)
??????????return?FALSE;
??
??????pWorkItem->UserProc?=?UserProc;
??????pWorkItem->UserParam?=?UserParam;
??
??????EnterCriticalSection(&pThreadPool->WorkItemLock);
??????InsertTailList(&pThreadPool->WorkItemHeader,?&pWorkItem->List);
??????LeaveCriticalSection(&pThreadPool->WorkItemLock);
??
??????InterlockedIncrement(&pThreadPool->WorkItemCount);
???????ReleaseSemaphore(pThreadPool->WorkItemSemaphore,?1,?NULL);
??
??????return?TRUE;
??}
??
??VOID?UserProc1(PVOID?dwParam)
??{
??????WorkItem(dwParam);
??}
??
??void?TestSimpleThreadPool(BOOL?bWaitMode,?LONG?ThreadNum)
??{
??????THREAD_POOL?ThreadPool;????
??????InitializeThreadPool(&ThreadPool,?ThreadNum);
??????
??????CompleteEvent?=?CreateEvent(NULL,?FALSE,?FALSE,?NULL);
??????BeginTime?=?GetTickCount();
??????ItemCount?=?20;
??
??????for(int?i=0;?i<20;?i++)
??????{
??????????PostWorkItem(&ThreadPool,?UserProc1,?(PVOID)bWaitMode);
??????}
??????
??????WaitForSingleObject(CompleteEvent,?INFINITE);
??????CloseHandle(CompleteEvent);
??
??????DestroyThreadPool(&ThreadPool);
??}
??????我們把工作項目放到一個隊列中，用一個信號量通知線程池，線程池中任意一個線程取出工作項目來執(zhí)行，執(zhí)行完畢之后，線程返回線程池，繼續(xù)等待新的工作項目。
??????線程池中線程的數(shù)量是固定的，預(yù)先創(chuàng)建好的，永久的線程，直到銷毀線程池的時候，這些線程才會被銷毀。
??????線程池中線程獲得工作項目的機會是均等的，隨機的，并沒有特別的方式保證哪一個線程具有特殊的優(yōu)先獲得工作項目的機會。
??????而且，同一時刻可以并發(fā)運行的線程數(shù)目沒有任何限定。事實上，在我們的執(zhí)行計算任務(wù)的演示代碼中，所有的線程都并發(fā)執(zhí)行。
??????下面，我們再來看一下，完成同樣的任務(wù)，系統(tǒng)提供的線程池是如何運作的。
??
??/**//************************************************************************/
??/**//*?QueueWorkItem?Test.??????????????????????????????????????????????????*/
??/**//************************************************************************/
??
??DWORD?BeginTime;
??LONG??ItemCount;
??HANDLE?CompleteEvent;
??
??int?compute()
??{
??????srand(BeginTime);
??
??????for(int?i=0;?i<20?*1000?*?1000;?i++)
??????????rand();
??
??????return?rand();
??}
??
??DWORD?WINAPI?WorkItem(LPVOID?lpParameter)
??{
??????BOOL?bWaitMode?=?(BOOL)lpParameter;
??
??????if(bWaitMode)
??????????Sleep(1000);
??????else
??????????compute();
??
??????if(InterlockedDecrement(&ItemCount)?==?0)
??????{
??????????printf("Time?total?%d?second.\n",?GetTickCount()?-?BeginTime);
??????????SetEvent(CompleteEvent);
??????}
??
??????return?0;
??}
??
??void?TestWorkItem(BOOL?bWaitMode,?DWORD?Flag)
??{
??????CompleteEvent?=?CreateEvent(NULL,?FALSE,?FALSE,?NULL);
??????BeginTime?=?GetTickCount();
??????ItemCount?=?20;
??????
??????for(int?i=0;?i<20;?i++)
??????{
??????????QueueUserWorkItem(WorkItem,?(PVOID)bWaitMode,?Flag);
??????}????
??
??????WaitForSingleObject(CompleteEvent,?INFINITE);
??????CloseHandle(CompleteEvent);
??}????? 很簡單，是吧？我們僅需要關(guān)注于我們的回調(diào)函數(shù)即可。但是與我們的簡單模擬來比，系統(tǒng)提供的線程池有著更多的優(yōu)點。
????? 首先，線程池中線程的數(shù)目是動態(tài)調(diào)整的，其次，線程池利用IO完成端口的特性，它可以限制并發(fā)運行的線程數(shù)目，默認(rèn)情況下，將會限制為CPU的數(shù)目，這可以減少線程切換。它挑選最近執(zhí)行過的線程再次投入執(zhí)行，從而避免了不必要的線程切換。
????? 系統(tǒng)提供的線程池背后的策略，我們下一節(jié)繼續(xù)再談。
?
? 參考書目
?
? 1，??? MSDN Library
? 2，??? 《Windows高級編程指南》
? 3，??? 《Windows核心編程》
? 4，??? 《Windows 2000 設(shè)備驅(qū)動程序設(shè)計指南》