線程池的研究及實現(xiàn)

1、什么是線程池?諸如web服務(wù)器、數(shù)據(jù)庫服務(wù)器、文件服務(wù)器和郵件服務(wù)器等許多服務(wù)器應(yīng)用都面向處理來自某些遠(yuǎn)程來源的大量短小的任務(wù)。構(gòu)建 服務(wù)器應(yīng)用程序的一個過于簡單的模型是：每當(dāng)一個請求到達(dá)就創(chuàng)建一個新的服務(wù)對彖，然后在新的服務(wù)對彖屮為請求服務(wù)。但當(dāng)有 大量請求并發(fā)訪問時，服務(wù)器不斷的創(chuàng)建和銷毀對彖的開銷很大。所以提高服務(wù)器效率的一個手段就是盡可能減少創(chuàng)建和銷毀對彖的 次數(shù)，特別是一些很耗資源的對彖創(chuàng)建和銷毀，這樣就引入了“池”的概念，“池”的概念使得人們可以定制一定量的資源，然后對這些資 源進(jìn)行復(fù)用，而不是頻繁的創(chuàng)建和銷毀。線程池是預(yù)先創(chuàng)建線程的一種技術(shù)。線程池在還沒有任務(wù)到來z前，創(chuàng)建一定數(shù)

2、量的線程，放入空閑隊列屮。這些線程都是處于睡眠 狀態(tài)，即均為啟動，不消耗cpu,而只是占用較小的內(nèi)存空間。當(dāng)請求到來z后，緩沖池給這次請求分配一個空閑線程，把請求傳入 此線程屮運(yùn)行，進(jìn)行處理。當(dāng)預(yù)先創(chuàng)建的線程都處于運(yùn)行狀態(tài)，即預(yù)制線程不夠，線程池可以自由創(chuàng)建一定數(shù)量的新線程，用于處理 更多的請求。當(dāng)系統(tǒng)比較閑的時候，也可以通過移除一部分一直處于停用狀態(tài)的線程。線程池的注意事項雖然線程池是構(gòu)建多線程應(yīng)用程序的強(qiáng)大機(jī)制，但使用它并不是沒有風(fēng)險的。在使用線程池時需注意線程池大小與性能的關(guān)系，注意 并發(fā)風(fēng)險、死鎖、資源不足和線程泄漏等問題。(1) 線程池大小。多線程應(yīng)用并非線程越多越好，需要根據(jù)系統(tǒng)運(yùn)

3、行的軟硬件環(huán)境以及應(yīng)用木身的特點決定線程池的大小。一般來說, 如果代碼結(jié)構(gòu)合理的話，線程數(shù)口與cpu數(shù)量相適合即可。如果線程運(yùn)行吋可能出現(xiàn)阻塞現(xiàn)象，可相應(yīng)增加池的大?。蝗缬斜匾?采用自適應(yīng)算法來動態(tài)調(diào)整線程池的大小，以提高cpu的右效利用率和系統(tǒng)的整體性能。(2) 并發(fā)錯誤。多線程應(yīng)用要特別注意并發(fā)錯誤，要從邏輯上保證程序的正確性，注意避免死鎖現(xiàn)象的發(fā)生。(3) 線程泄漏。這是線程池應(yīng)用屮一個嚴(yán)重的問題，當(dāng)任務(wù)執(zhí)行完畢而線程沒能返冋池屮就會發(fā)生線程泄漏現(xiàn)象。簡單線程池的設(shè)計一個典型的線程池，應(yīng)該包括如下幾個部分：1、線程池管理器(threadpool),用于啟動、停用，管理線程池2、工作線程

4、(workthread),線程池中的線程3、請求接口(workrequest),創(chuàng)建請求對象，以供工作線程調(diào)度任務(wù)的執(zhí)行4、請求隊列(requestqueue)，用于存放和提取請求5、結(jié)杲隊列(resultqueue),用于存儲請求執(zhí)行后返回的結(jié)杲 線程池管理器，通過添加請求的方法(putrequest)向請求隊列(requestqueue)添加請求，這些請求事先需要實現(xiàn)請求接口，即 傳遞工作函數(shù)、參數(shù)、結(jié)果處理函數(shù)、以及異常處理函數(shù)。z后初始化一定數(shù)量的工作線程，這些線程通過輪詢的方式不斷查看請求 隊列(requestqueue),只要有請求存在，則會提取出請求，進(jìn)行執(zhí)行。然后，線程池管理器

5、調(diào)用方法(poll)查看結(jié)果隊列(resultqueue) 是否有值，如果有值，則取出，調(diào)用結(jié)果處理函數(shù)執(zhí)行。通過以上講述，不難發(fā)現(xiàn)，這個系統(tǒng)的核心資源在于請求隊列和結(jié)果隊列， 工作線程通過輪詢requestqueue獲得人物，主線程通過查看結(jié)果隊列，獲得執(zhí)行結(jié)果。因此，對這個隊列的設(shè)計，要實現(xiàn)線程同步, 以及一定阻塞和超時機(jī)制的設(shè)計，以防止因為不斷輪詢而導(dǎo)致的過多cpu開銷。在本文屮，將會用python語言實現(xiàn)，python的queue, 就是很好的實現(xiàn)了對線程同步機(jī)制。使用python實現(xiàn):#-*-encoding:utf-8-*-，)，created on 2012-3-9©s

6、ummary:線程池contact: mai1to:zhanglixinscugmai1. comauthor: zhanglixin，)，import sysimport threadingimport queueimport traccback#定義一些exception,用于口定義異常處理class norcsultspcnding(exccption):"""all works requests have been processedpass class noworkersavailablc(exccption):worket threads availa

7、ble to process rcmaining requests.pass def _handle_thread_exception(request, exc_info):默認(rèn)的異常處理函數(shù)，只是簡單的打印"traccback. print cxccption(*cxc info)ttclasscsclass workcrthrcad(threading. thread):""后臺線程，真正的工作線程，從請求隊列(rcqucstqucuc)屮獲取work, 并將執(zhí)行后的結(jié)果添加到結(jié)果隊列(rcsultqueue)/，/，/，def init (self，rcqu

8、cstqucuc, resultqueue, poll timcout=5, *kwds): threading. thread. _init_(self, *kwds)'''設(shè)置為守護(hù)進(jìn)行self, setdacmon(true)self. _requcstqueue 二 requestqueueself. _resultqucue 二 resuitqueueself. _poll_timcout 二 poll_timeout'''設(shè)置一個flag信號，用來表示該線程是否還被dismiss,默認(rèn)為false''' sel

9、f, -dismissed 二 threading. event()self, st art ()def run(self):'''每個線程盡口j能多的執(zhí)行work,所以采用loop, 只要線程叫用，并il requestqueue有work未完成，則一直loop''' whi1e true:if self, dismissed, is set ():breakt ry:，queue. queue隊列設(shè)置了線程同步策略，并且可以設(shè)置timcouto 一直block,直到rcqucstqucuc有值，或者超時，request = self. rcq

10、ucstqucuc. get(true, self. _poll timeout)except queue. empty:continueelse:'''之所以在這里再次判斷dimissed,是因為之前的timcout時間里，很有可能，該線程被dismiss掉了'' if self.-dismissed. is_set():self. _requestqucuc. put(request)breaktry:'''執(zhí)彳亍callable,講請求和結(jié)果以tuple的方式放入rcqucstqucuc，''resuit 二

11、 request.cedlablc(*request. args, *request. kwds)print self, gctname ()self. _resuitqueue, put(request, resuit)except:'''異常處理'''request. exception 二 trueself. rcsultqucuc. put (request, sys. cxc info ()def dismiss(self):'''設(shè)置一個標(biāo)志，表示完成當(dāng)前work之后，退出self, dismissed, se

12、t ()class workrequest:，)，param callable.:,可定制的,執(zhí)行work的函數(shù)param args:列表參數(shù)param kwds:字典參數(shù)param requestid: idparam callback:可定制的，處理rcsultqucuc隊列元素的函數(shù)param exc_callback:可定制的,處理異常的函數(shù),def _init_(self, callable_, args=none，kwds二none, requestld=none, callback=none, cxc_callback=_handle_thrcad_cxccption): if

13、requestid 二二 none:self, requestid 二 id(self)el se:t ry:self.requestid = hash (requesttd)except typcerror:raise typcerror("rcqucstld must be hashable")self, exception 二 falseself, callback = callbackself, exc callback = cxc callbackself, callable = callableself, args = args or self, kwds =

14、kwds or def _str_(self):return z，workrcqucst id=%s args=%r kwargs=%r exception二s % (self, requestid，self, args, self, kwds, self, cxccption)class thrcadpool:,param numworkers:初始化的線程數(shù)量param q_sizc, rcsq_sizc: rcqucstqucuc 和 result 隊列的初始大小param poll_timcout:設(shè)置工作線程 workcrthrcad 的 timeout,也就是等待 rcqucstq

15、ucuc 的 timcout ,def _init_(self, num_workers, q_size=0, resq_size=o, poll_timoout=5):self. _requestqueue 二 queue. queue(q_size)self. _resultqueue 二 queue. queue(resq_size)seif. workers 二self, dismissedworkcrs =self. workrcquests = #設(shè)置個字典，方便使用self, crcatcworkers(num workers, poll timeout)def crcatcwo

16、rkcrs(self, num workers, poll timcout=5):'''創(chuàng)建 num workers 個 workthread,默認(rèn) timeout 為 5'''for i in range(num workers):self, workers, append(workerthread(self. rcqucstqucuc, self. resultqueue, poll timcout二poll timeout)def dismissworkers(self, num_workers, do_join=falsc):'&

17、#39;'停用num_workcrs數(shù)量的線程,并加入dismiss_list，''dismiss_list 二for i in range(min(num_workcrs, len(self, workers):worker = self, workers. pop()worker, dismiss ()dismiss_list. append(worker)if do_join :for worker in dismiss_list:worker. join()else:seif. di smissedworkers. extend(dismiss_list)def

18、 joinalidismissedworkers(seif):"'join所有停用的thread'"#print lcn(sclf dismissedworkers)for worker in self, dismissedworkers:worker. join()self, dismissedworkers =def putrequest(self，request , block=truc, timcout=nonc):asscrt isinstancc(request，workrequest)assert not getattr (request,&#

19、39; exception，, none)'''當(dāng)queue滿了,也就是容量達(dá)到了前面設(shè)定的q_sizc,它將一直阻塞,直到有空余位置,或是timeout'' self. _requestqucue. put(request, block, timcout)self. workrequestsrcqucst. requestid二 requestdef poll (self, block = false):while true:if not self. workrequests:raise norcsuitspendingelif block emd n

20、ot self, workers:raise noworkersavailablctry:'''默認(rèn)只要resultqueue有值,則取出，否則一直block'''request , resuit 二 self. _resuitqueue, get(block二block)if request. exception and request. exc_caliback:request, cxc callback (request, result)callback):if request, callback and not (request, exc

21、eption and request.cxc request.callback (request, result)del self. workrequestsrequest. requestidexcept queue. empty:breakdef wait(self):while true:try:seif poll (true)except norcsuitspending:breakdef workersize(self):return lcn(sclf. workers)def stop(self):'''join所有的thread,確保所有的線程都執(zhí)行完畢&

22、#39;self. dismissworkers(self, workersize(), true)self.joinal 1 di smi ssedworkers()測試代碼:#test a demoif _name_=' _meiin_' :import randomimport timeimport dcitetimedef do_work(data):time, sleep(random. nmdint (1，3)res = str (date ti me. datetime .no w () + "” +st r (data)return resdef pr

23、int_result(request, result):print "resuit from request %s : %r % (request. requestid, result)main = threadpool(3)for i in range(40):req = workrequest(do_work, args=i, kwds二，callback=print_result) main. putrequest(req)print "work request #%s added.，z % req. requesltdprint*20，main, workersize 0 /*20counter = 0while true:t ry:ti