線程池
(本章節中例子都是用 VS2010 編譯調試的)
線程池編寫必須在 Windows Vista 操作系統(以及以上版本的操作系統)下,且 C++ 編譯器版本至少是 VS2008
線程池的功能
- 以異步的方式來調用一個函數
- 每隔一段時間調用一個函數
- 當內核對象觸發的時候調用一個函數
- 當異步 I/O 請求完成的時候調用一個函數
注意
當一個進程初始化的時候,它並沒有任何與線程池的開銷.但是,一旦調用了新的線程池函數,系統就會為進程相應的內核資源,其中的一些資源在進程終止之前都將一直存在.正如我們可以看到,使用線程池的開銷取決於用法:系統會以進程的名義來分配線程,其他內核以及內部數據結構.因此我們不應該盲目地使用這些線程池函數,而是必須謹慎地考慮,這些函數能做什么,以及它們不能做什么.
在線程池編程中,我們從來不需要自己調用 CreateThread.系統會自動為我們的進程創建線程,並在規定的條件下讓線程池中的線程調用我們的回調函數.此外,這個線程在處理完成一個客戶請求后,它不會立刻被銷毀,而是回到線程池,准備好處理隊列中的任何其他工作項,線程池會不斷地重復使用其中的線程,而不會頻繁地創建銷毀線程,對應用程序來說,這樣可以顯著地提升性能,因為創建和銷毀線程會消耗大量的時間.當然,如果線程池檢測到創建的另一個線程將能更好地為應用程序服務,那么它會這樣做.如果線程池檢測到它的線程數量已經供過於求,那么它會銷毀其中一些線程.除非我們非常清楚自己在做什么,否則的話最好還是相信線程內部的算法,讓它自動地對應用程序的工作量進行微調.
默認線程池,在進程存在期間它不會被銷毀.生命周期與進程相同.在進程終止的時候,Windows 會將其銷毀並負責所有的清理工作.
對線程池的制定
可以用 CreateThreadpool 來創建一個新的線程池,該函數返回一個 PTP_POOL 值,表示新創建的線程池.接着我們可以調用后面兩個函數來設置線程池中線程的最大數量和最小數量 SetThreadpoolThreadMinimum / SetThreadpoolThreadMaximum 線程池始終保持池中的線程數量至少是指定的最小數量,並允許線程數量增長到指定的最大數量,順便一提,默認線程池的最小數量為1,最大數量為500.然后在引用程序不在需要自定義線程池時,應調用 CloseThreadpool 將其銷毀.在調用這個函數后,我們將無法在將任何新的項添加到線程池的隊列中.線程池中當前正在處理的隊列中的線程會完成它們的處理並終止.此外,線程池的隊列中所有尚未開始處理的項將被取消.
一旦我們創建了自己的線程池,並制定了線程池的最小數量和最大數量,我們就可以初始化一個回調環境,它包含了一些課應用於工作項的額外的設置或配置.(線程池回調環境結構 _TP_CALLBACK_ENVIRON ,其定義在 WinNT.h )
View Code
/*************************************** 顯示此結果的編譯環境為 VS2010 ***************************************/ #if (_WIN32_WINNT >= _WIN32_WINNT_WIN7) typedef struct _TP_CALLBACK_ENVIRON_V3 { TP_VERSION Version; PTP_POOL Pool; PTP_CLEANUP_GROUP CleanupGroup; PTP_CLEANUP_GROUP_CANCEL_CALLBACK CleanupGroupCancelCallback; PVOID RaceDll; struct _ACTIVATION_CONTEXT *ActivationContext; PTP_SIMPLE_CALLBACK FinalizationCallback; union { DWORD Flags; struct { DWORD LongFunction : 1; DWORD Persistent : 1; DWORD Private : 30; } s; } u; TP_CALLBACK_PRIORITY CallbackPriority; DWORD Size; } TP_CALLBACK_ENVIRON_V3; typedef TP_CALLBACK_ENVIRON_V3 TP_CALLBACK_ENVIRON, *PTP_CALLBACK_ENVIRON; #else typedef struct _TP_CALLBACK_ENVIRON_V1 { TP_VERSION Version; PTP_POOL Pool; PTP_CLEANUP_GROUP CleanupGroup; PTP_CLEANUP_GROUP_CANCEL_CALLBACK CleanupGroupCancelCallback; PVOID RaceDll; struct _ACTIVATION_CONTEXT *ActivationContext; PTP_SIMPLE_CALLBACK FinalizationCallback; union { DWORD Flags; struct { DWORD LongFunction : 1; DWORD Persistent : 1; DWORD Private : 30; } s; } u; } TP_CALLBACK_ENVIRON_V1; typedef TP_CALLBACK_ENVIRON_V1 TP_CALLBACK_ENVIRON, *PTP_CALLBACK_ENVIRON; #endif
然后我們可以調用 InitializeThreadpoolEnvironment 初始化這個結構體,接着當然回調環境必須調用 SetThreadpoolCallbackPool 標明給工作項應該由哪個線程池來處理.最后當我們不在需要這個使用線程池回調環境的時候,應該調用 DestroyThreadpoolEnvironment 來對它進行清理工作.
-
- 然而可以調用 SetThreadpoolCallbackRunsLong 函數來告訴回調環境,工作項通常需要較長的時間來處理.這使得線程池會更快地創建線程,其目的是為了嘗試在對工作項進行處理的時候,以一種更為公平的方式來替代最有效的方式.
- 也可以調用 SetThreadpoolCallbackLibrary 來確保只要線程池中還有待處理的工作項,就將一個特定的 DLL 一直保持在進程空間中.基本上 SetThreadpoolCallbackLibrary 函數的存在目的是為了消除潛在的競態條件(race condition),從而避免可能導致死鎖.這個相當高級的特性,更詳細信息參閱 Platform SDK 文檔.
線程池的銷毀(清理組)
為了得體地銷毀私有線程池,我們首先可以需要通過調用 CreateThreadpoolCleanupGroup 來創建一個清理組,然后再將這個清理組與一個以綁定到線程池的回調函數結構體 TP_CALLBACK_ENVIRON 調用 SetThreadpoolCallbackCleanupGroup 函數把兩者關聯起來.其中 SetThreadpoolCallbackCleanupGroup 的 pfng 參數標識一個回調函數的地址(函數原型 即CleanupGroupCancelCallback),如果傳給 pfng 參數值不為 NULL ,且當清理組被取消時那么這個回調函數會被調用.
當我們調用 CreateThreadpoolWork, CreateThreadpoolTimer, CreateThreadpoolWait 或 CreateThreadpoolIo 的時候,如果最后那個參數,即指向 PTP_CALLBACK_ENVIRON 結構體指針,不等於 NULL,那么所創建的項會被添加到對應的回調環境的清理組中,其目的是為了表示有線程池中添加了一項,需要潛在清理.在這些對了項完成后,如果我們調用 CloseThreadpoolWork, CloseThreadpoolTimer, CloseThreadpoolWait 和 CloseThreadpoolIo, 那就等於是隱式將對應的項從組中移除.
最后,在程序想要銷毀線程池的時候,調用 CloseThreadpoolCleanupGroupMembers.這個函數與下面的 WaitForThreadpool*(例: WaitForThreadpoolWork)函數相似.當線程調用 CloseThreadpoolCleanupGroupMembers 時候,函數會一直等待,知道線程池的工作組中所有剩余的項(即已經創建當尚未關閉的項)都已經處理完畢為止.調用者還可以傳 TRUE 給 fCancelPendingCallbacks 參數.這樣會將所有已提交但尚未處理的工作項直接取消,函數會在所有當前正在運行的工作項王城之后返回.如果傳給 fCancelPendingCallbacks 參數為 TRUE,而且傳給 SetThreadpoolCallbackCleanupGroup 的 pfng 參數值是一個 CleanupGroupCancelCallback 函數地址,那么對每一個被取消的工作項,我們的回調函數會被調用,在 CleanupGroupCancelCallback 函數中.參數 ObjectContext 會包含每個被取消的上下文.(該上下文信息是通過 CreateThreadpool* 函數的 pv 參數設置的)在 CleanupGroupCancelCallback 函數中, 參數 CleanupContext 包含的上下文是通過 CloseThreadpoolCleanupGroupMembers 函數的 pvCleanupContext 參數傳入的.如果在調用 CloseThreadpoolCleanupGroupMembers 時傳入 FALSE 給 fCancelPendingCallbacks 參數.那么在返回之前,線程池會發時間來處理隊列中所有剩余的項.注意這種情況下我們的 CleanupGroupCancelCallback 函數絕對不會被調用.因此可以傳 NULL 給 pvCleanupContext 參數.
線程池制定與銷毀代碼樣例
編寫步驟
程序源碼
View Code
#include<Windows.h> #include<iostream> #include<cstdlib> using namespace std; VOID CALLBACK SimpleCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context); void main() { PTP_POOL tPool; TP_CALLBACK_ENVIRON pcbe; //創建線程池 tPool = CreateThreadpool(NULL); //設置線程池最大最小的線程數量 SetThreadpoolThreadMinimum(tPool,1); SetThreadpoolThreadMaximum(tPool,2); //初始化線程池環境變量 InitializeThreadpoolEnvironment(&pcbe); //為線程池設置線程池環境變量 SetThreadpoolCallbackPool(&pcbe,tPool); //單次工作提交 TrySubmitThreadpoolCallback(SimpleCallback,NULL,&pcbe); system("pause"); //清理線程池的環境變量 DestroyThreadpoolEnvironment(&pcbe); //關閉線程池 CloseThreadpool(tPool); } VOID CALLBACK SimpleCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context) { cout<<"this is SimpleCallback function!"<<endl; }
運行結果
以異步的方式調用函數
相關函數
- TrySubmitThreadpoolCallback (向線程池提交工作請求函數<一次性>,回調函數原型)
- CreateThreadpoolWork (為線程池創建一個提交工作的工作對象,回調函數原型)
絕對不要讓回調函數調用 WaitForThreadpoolWork 並將自己的工作項作為參數傳入,因為這樣會導致死鎖. - SubmitThreadpoolWork (想線程池提交工作請求函數<非一次性>)
- WaitForThreadpoolWorkCallbacks (取消已提交但未執行的工作項 / 等待工作項處理完成把自己掛起)
其中的 fCancelPendingCallbacks 參數
若為 true ,函數會試圖取消先前提交的那個工作項.如果線程池中的線程正在處理那個工作項,那么該過程不會被打斷,函數會一直等到該工作項已經被處理完成后在返回.
若為 false ,那么函數會將調用線程掛起,知道指定工作項的處理已經完成而且線程池中處理該工作項的線程也已經被回收並准備處理下一個工作項.
如果用一個 PTP_WORK 對象已經提交了很多工作項,而且傳給 fCancelPendingCallbacks 參數為 false,那么 WaitForThreadpoolWorkCallbacks 會等待線程池處理完所有提交的工作項.如果傳給 fCancelPendingCallbacks 為 true,那么 WaitForThreadpoolWorkCallbacks 只會等到當前正在運行的工作項完成為止. - CloseThreadpoolWork (取消可以多次提交工作的工作對象)
編寫步驟
代碼樣例
程序源碼
View Code
#include<Windows.h> #include<iostream> #include<cstdlib> using namespace std; VOID CALLBACK SimpleCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context); VOID CALLBACK WorkCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context,PTP_WORK Work); void main() { PTP_WORK tpWork; //單次工作提交 TrySubmitThreadpoolCallback(SimpleCallback,NULL,NULL); //創建工作對象 tpWork = CreateThreadpoolWork(WorkCallback,NULL,NULL); //提交工作 SubmitThreadpoolWork(tpWork); SubmitThreadpoolWork(tpWork); //等待工作結束 WaitForThreadpoolWorkCallbacks(tpWork,false); //關閉工作對象 CloseThreadpoolWork(tpWork); system("pause"); } VOID CALLBACK SimpleCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context) { cout<<"this is SimpleCallback function!"<<endl; } VOID CALLBACK WorkCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context,PTP_WORK Work) { cout<<"this is WorkCallback function!"<<endl; }
運行結果
以時間段來調用函數
相關函數
- CreateThreadpoolTimer (為線程池創建一個定時調用的工作對象,回調函數原型)
絕對不要讓回調函數調用 WaitForThreadpoolTimer 並將自己的工作項作為參數傳入,因為這樣會導致死鎖. - SetThreadpoolTimer (為線程池的定時調用工作對象注冊計時器,也可以修改以注冊的定時調用工作對象的計時器,調用方式有些類似可等待計時器對象)
- WaitForThreadpoolTimerCallbacks (取消已提交但未執行的定時工作項 / 等待定時工作項處理完成把自己掛起)
- CloseThreadpoolTimer (取消定時調用的工作對象)
- IsThreadpoolTimerSet (判斷定時調用的工作對象是否設置過計時器)
編寫步驟
代碼樣例
程序源碼
View Code
#include<Windows.h> #include<iostream> #include<cstdlib> using namespace std; VOID CALLBACK TimerCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context,PTP_TIMER Timer); void main() { PTP_TIMER tpTimer; FILETIME liDueTime; LARGE_INTEGER liUTC; liUTC.QuadPart = -30000000; liDueTime.dwLowDateTime = liUTC.LowPart; liDueTime.dwHighDateTime = liUTC.HighPart; //創建定時調用的工作對象 tpTimer = CreateThreadpoolTimer(TimerCallback,NULL,NULL); //判斷定時調用的工作對象是否注冊過計時器 if(!IsThreadpoolTimerSet(tpTimer)) { //為定時調用工作對象注冊計時器 SetThreadpoolTimer(tpTimer,&liDueTime,0,0); } //睡眠主進程,等待計時器添加工作 Sleep(4000); //等待定時調用的工作對象 WaitForThreadpoolTimerCallbacks(tpTimer,false); //關閉定時調用的工作對象 CloseThreadpoolTimer(tpTimer); system("pause"); } VOID CALLBACK TimerCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context,PTP_TIMER Timer) { cout<<"this is TimerCallback function!"<<endl; }
運行結果
以內核對象的觸發狀態來調用函數
運行原理
其實,在此功能中線程池在內部會讓一個線程調用 WaitForMultipleObjects 函數,傳入通過 SetThreadpoolWait 函數注冊一組句柄,並傳入 false 給 bWaitAll 參數,這樣當任何一個句柄被觸發的時候,線程池就會被喚醒.由於 WaitForMultipleObjects 有一個限制,一次嘴甜只能等待 64(MAXMUM_WAIT_OBJECTS)個句柄,因此線程池事實上正是為每 64 個內核對象分配一個線程來進行等待,其效率還是相當高的.另外,由於 WaitForMultipleObjects 不允許我們將同一個句柄傳入多次,因此我們必須確保不會用 SetThreadpoolWait 來多次注冊同一個句柄.但是,我們可以調用 DuplicateHandle, 這樣就可以分別注冊原始句柄和復制句柄.
注意
絕對不要讓回調函數調用 WaitForThreadpoolWait 並將自己的工作項作為參數傳入,因為這樣會導致死鎖.另外,當線程在等待傳給 SetThreadpoolWait 的句柄時,我們應該確保該句柄不會被關閉.最后我們可能並不想通過 PulseEvent 來觸發一個已注冊的事件,因為當 PulseEvent 被調用的時候,我們無法保證線程池正好在等待該事件.
相關函數
- CreateThreadpoolWait (為線程池創建一個等待內核對象觸發的工作對象,回調函數原型)
- SetThreadpoolWait (將內核綁定到等待內核對象觸發的工作對象上)
如果想讓回調函數在同一個內核對象被觸發的時候再次被調用,那么需要調用 SetThreadpoolWait 來再次注冊.也可以通過 SetThreadpoolWait 來重用該等待項,即可以傳入一個不同的內核對象句柄,也可以傳入 NULL 來將該等待項從線程池中移除 - WaitForThreadpoolWaitCallbacks (取消已提交但未執行的等待內核對象觸發的工作項 / 等待現有等待內核對象觸發的工作項處理完成把自己掛起)
- CloseThreadpoolWait (取消等待內核對象觸發的工作對象)
編寫步驟
代碼樣例
程序源碼
View Code
#include<Windows.h> #include<iostream> #include<cstdlib> using namespace std; VOID CALLBACK WaitCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context,PTP_WAIT Wait,TP_WAIT_RESULT WaitResult); void main() { PTP_WAIT tpWait; HANDLE hMutex; //創建等待內核對象觸發的工作對象 tpWait = CreateThreadpoolWait(WaitCallback,NULL,NULL); //創建互斥對象 hMutex=CreateMutex(NULL,false,0); //等待互斥對象被釋放后獲得互斥對象擁有權 WaitForSingleObject(hMutex,INFINITE); //設置等待內核事件觸發調用回調函數 SetThreadpoolWait(tpWait,hMutex,0); //釋放互斥對象擁有權 ReleaseMutex(hMutex); //等待等待內核對象觸發的工作對象結束 WaitForThreadpoolWaitCallbacks(tpWait,false); //關閉等待內核對象觸發的工作對象 CloseThreadpoolWait(tpWait); system("pause"); } VOID CALLBACK WaitCallback(PTP_CALLBACK_INSTANCE Instance,PVOID Context,PTP_WAIT Wait,TP_WAIT_RESULT WaitResult) { cout<<"this is WaitCallback function!"<<endl; }
運行結果
以異步 I/O 請求完成時調用函數
相關函數
- CreateThreadpoolIo (為線程池創建一個異步 I/O 請求完成的工作對象,回調函數原型)
- StartThreadpoolIo (將嵌入在 I/O 項中的文件/設備與線程池內部的 I/O 完成端口關聯起來)
在每次調用 ReadFile 和 WriteFile 之前,我們必須調用 StartThreadpoolIo.如果每次在發出 I/O 請求之前沒有調用 StartThreadpoolIo,那么 IoCompletionCallback 回調函數將不會被回調 - CancelThreadpoolIo (在發出 I/O 請求之后讓線程池停止調用回調函數)
- WaitForThreadpoolIoCallbacks (取消已提交但未執行的異步 I/O 請求完成的工作項 / 等待異步 I/O 請求完成的工作項處理完成把自己掛起)
- CloseThreadpoolIo (取消異步 I/O 請求完成的工作對象)
代碼樣例
程序源碼
運行結果
對回調函數的操作
回調函數的終止操作
線程池提供了一種便利的方法,用來描述在我們的回調函數返回之后,應該執行的一些操作,回調函數用傳給它的不透明的 Instance 參數(其類型為 PTP_CALLBACK_INSTANCE)來調用下面的函數
- LeaveCriticalSectionWhenCallbackReturns (當回調函數返回的時候,線程池會自動調用 LeaveCriticalSection,並在參數中傳入指定的 CRITICAL_SECTION 結構)
- ReleaseMutexWhenCallbackReturns (當回調函數返回的時候,線程池會自動調用 ReleaseMutex ,並在參數傳入指定的 HANDLE)
- ReleaseSemaphoreWhenCallbackReturns (當回調函數返回的時候,線程池會自動調用 ReleaseSemaphore,並在參數傳入指定的 HANDLE)
- SetEventWhenCallbackReturns (當回調函數返回的時候,線程池會自動調用 SetEvent,並在參數中傳入指定的 HANDLE)
- FreeLibraryWhenCallbackReturns (當回調函數返回的時候,線程池會自動調用 FreeLibrary, 並在參數中傳入指定的 HMODULE)
另外兩個函數
- CallbackMayRunLong
此函數用來通知線程池回調函數的運行時間會比較長.如果一個回調函數認為自己需要較長的時間來處理當前的項,那么它應該調用 CallbackMayRunLong.由於線程池會堅持不創建新線程,因為長時間運行的項可能會使線程池隊列中的其他項挨餓.如果 CallbackMayRunLong 返回 TRUE,那么說明線程池中還有其他線程可供使用,來對隊列中的項進行處理.如果 CallbackMayRunLong 返回 false,那么說明線程池中沒有其他線程可以用來處理隊列中的項.為了維持線程池的運行效率,最好是讓該項將它的任務划分成更小的步伐來處理(將每一個部分單獨地添加到線程池的隊列中).任務的第一部分可以當前線程中執行. - DisassociateCurrentThreadFromCallback
回調函數調用它來告訴線程池,邏輯上自己已經完成了工作,這使得任何由於調用 WaitForThreadpoolWorkCallbacks, WaitForThreadpoolTimerCallbacks, WaitForThreadpoolWaitCallbacks 或 WaitForThreadpoolIoCallbacks 而被阻塞的線程能夠早一些返回,而不必等到線程池的線程從回調函數中返回