Python之ThreadPoolExecutor線程池問題
概念
Python
中已經(jīng)有了threading
模塊,為什么還需要線程池呢,線程池又是什么東西呢?
以爬蟲為例,需要控制同時爬取的線程數(shù),例子中創(chuàng)建了20個線程,而同時只允許3個線程在運行,但是20個線程都需要創(chuàng)建和銷毀,線程的創(chuàng)建是需要消耗系統(tǒng)資源的,有沒有更好的方案呢?
其實只需要三個線程就行了,每個線程各分配一個任務(wù),剩下的任務(wù)排隊等待,當(dāng)某個線程完成了任務(wù)的時候,排隊任務(wù)就可以安排給這個線程繼續(xù)執(zhí)行。
這就是線程池的思想(當(dāng)然沒這么簡單),但是自己編寫線程池很難寫的比較完美,還需要考慮復(fù)雜情況下的線程同步,很容易發(fā)生死鎖。
從Python3.2
開始,標(biāo)準(zhǔn)庫為我們提供了concurrent.futures
模塊,它提供了ThreadPoolExecutor
和ProcessPoolExecutor
兩個類,實現(xiàn)了對threading
和multiprocessing
的進一步抽象(這里主要關(guān)注線程池),不僅可以幫我們自動調(diào)度線程,還可以做到:
- 主線程可以獲取某一個線程(或者任務(wù)的)的狀態(tài),以及返回值。
- 當(dāng)一個線程完成的時候,主線程能夠立即知道。
- 讓多線程和多進程的編碼接口一致。
實例
簡單使用
from concurrent.futures import ThreadPoolExecutor import time # 參數(shù)times用來模擬網(wǎng)絡(luò)請求的時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times executor = ThreadPoolExecutor(max_workers=2) # 通過submit函數(shù)提交執(zhí)行的函數(shù)到線程池中,submit函數(shù)立即返回,不阻塞 task1 = executor.submit(get_html, (3)) task2 = executor.submit(get_html, (2)) # done方法用于判定某個任務(wù)是否完成 print(task1.done()) # cancel方法用于取消某個任務(wù),該任務(wù)沒有放入線程池中才能取消成功 print(task2.cancel()) time.sleep(4) print(task1.done()) # result方法可以獲取task的執(zhí)行結(jié)果 print(task1.result()) # 執(zhí)行結(jié)果 # False # 表明task1未執(zhí)行完成 # False # 表明task2取消失敗,因為已經(jīng)放入了線程池中 # get page 2s finished # get page 3s finished # True # 由于在get page 3s finished之后才打印,所以此時task1必然完成了 # 3 # 得到task1的任務(wù)返回值
ThreadPoolExecutor構(gòu)造實例的時候,傳入max_workers參數(shù)來設(shè)置線程池中最多能同時運行的線程數(shù)目。
使用submit函數(shù)來提交線程需要執(zhí)行的任務(wù)(函數(shù)名和參數(shù))到線程池中,并返回該任務(wù)的句柄(類似于文件、畫圖),注意submit()不是阻塞的,而是立即返回。
通過submit函數(shù)返回的任務(wù)句柄,能夠使用done()方法判斷該任務(wù)是否結(jié)束。上面的例子可以看出,由于任務(wù)有2s的延時,在task1提交后立刻判斷,task1還未完成,而在延時4s之后判斷,task1就完成了。
使用cancel()方法可以取消提交的任務(wù),如果任務(wù)已經(jīng)在線程池中運行了,就取消不了。這個例子中,線程池的大小設(shè)置為2,任務(wù)已經(jīng)在運行了,所以取消失敗。如果改變線程池的大小為1,那么先提交的是task1,task2還在排隊等候,這是時候就可以成功取消。
使用result()方法可以獲取任務(wù)的返回值。查看內(nèi)部代碼,發(fā)現(xiàn)這個方法是阻塞的。
as_completed
上面雖然提供了判斷任務(wù)是否結(jié)束的方法,但是不能在主線程中一直判斷啊。
有時候我們是得知某個任務(wù)結(jié)束了,就去獲取結(jié)果,而不是一直判斷每個任務(wù)有沒有結(jié)束。
這是就可以使用as_completed
方法一次取出所有任務(wù)的結(jié)果。
from concurrent.futures import ThreadPoolExecutor, as_completed import time # 參數(shù)times用來模擬網(wǎng)絡(luò)請求的時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times executor = ThreadPoolExecutor(max_workers=2) urls = [3, 2, 4] # 并不是真的url all_task = [executor.submit(get_html, (url)) for url in urls] for future in as_completed(all_task): data = future.result() print("in main: get page {}s success".format(data)) # 執(zhí)行結(jié)果 # get page 2s finished # in main: get page 2s success # get page 3s finished # in main: get page 3s success # get page 4s finished # in main: get page 4s success
as_completed()
方法是一個生成器,在沒有任務(wù)完成的時候,會阻塞,在有某個任務(wù)完成的時候,會yield
這個任務(wù),就能執(zhí)行for循環(huán)下面的語句,然后繼續(xù)阻塞住,循環(huán)到所有的任務(wù)結(jié)束。
從結(jié)果也可以看出,先完成的任務(wù)會先通知主線程。
map
除了上面的as_completed
方法,還可以使用executor.map
方法,但是有一點不同。
from concurrent.futures import ThreadPoolExecutor import time # 參數(shù)times用來模擬網(wǎng)絡(luò)請求的時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times executor = ThreadPoolExecutor(max_workers=2) urls = [3, 2, 4] # 并不是真的url for data in executor.map(get_html, urls): print("in main: get page {}s success".format(data)) # 執(zhí)行結(jié)果 # get page 2s finished # get page 3s finished # in main: get page 3s success # in main: get page 2s success # get page 4s finished # in main: get page 4s success
使用map
方法,無需提前使用submit
方法,map
方法與python
標(biāo)準(zhǔn)庫中的map
含義相同,都是將序列中的每個元素都執(zhí)行同一個函數(shù)。
上面的代碼就是對urls
的每個元素都執(zhí)行get_html
函數(shù),并分配各線程池。可以看到執(zhí)行結(jié)果與上面的as_completed
方法的結(jié)果不同,輸出順序和urls
列表的順序相同,就算2s的任務(wù)先執(zhí)行完成,也會先打印出3s的任務(wù)先完成,再打印2s的任務(wù)完成。
wait
wait
方法可以讓主線程阻塞,直到滿足設(shè)定的要求。
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED import time # 參數(shù)times用來模擬網(wǎng)絡(luò)請求的時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times executor = ThreadPoolExecutor(max_workers=2) urls = [3, 2, 4] # 并不是真的url all_task = [executor.submit(get_html, (url)) for url in urls] wait(all_task, return_when=ALL_COMPLETED) print("main") # 執(zhí)行結(jié)果 # get page 2s finished # get page 3s finished # get page 4s finished # main
wait
方法接收3個參數(shù),等待的任務(wù)序列、超時時間以及等待條件。
等待條件return_when
默認(rèn)為ALL_COMPLETED
,表明要等待所有的任務(wù)都結(jié)束。
可以看到運行結(jié)果中,確實是所有任務(wù)都完成了,主線程才打印出main
。
等待條件還可以設(shè)置為FIRST_COMPLETED
,表示第一個任務(wù)完成就停止等待。
源碼分析
cocurrent.future
模塊中的future
的意思是未來對象,可以把它理解為一個在未來完成的操作,這是異步編程的基礎(chǔ) 。
在線程池submit()
之后,返回的就是這個future
對象,返回的時候任務(wù)并沒有完成,但會在將來完成。
也可以稱之為task的返回容器,這個里面會存儲task的結(jié)果和狀態(tài)。
那ThreadPoolExecutor
內(nèi)部是如何操作這個對象的呢?
下面簡單介紹ThreadPoolExecutor
的部分代碼:
1.init方法
init
方法中主要重要的就是任務(wù)隊列和線程集合,在其他方法中需要使用到。
2.submit方法
submit
中有兩個重要的對象,_base.Future()
和_WorkItem()
對象,_WorkItem()
對象負(fù)責(zé)運行任務(wù)和對future
對象進行設(shè)置,最后會將future
對象返回,可以看到整個過程是立即返回的,沒有阻塞。
3.adjust_thread_count方法
這個方法的含義很好理解,主要是創(chuàng)建指定的線程數(shù)。但是實現(xiàn)上有點難以理解,比如線程執(zhí)行函數(shù)中的weakref.ref,涉及到了弱引用等概念,留待以后理解。
4._WorkItem對象
_WorkItem
對象的職責(zé)就是執(zhí)行任務(wù)和設(shè)置結(jié)果。這里面主要復(fù)雜的還是self.future.set_result(result)
。
5.線程執(zhí)行函數(shù)--_worker
這是線程池創(chuàng)建線程時指定的函數(shù)入口,主要是從隊列中依次取出task執(zhí)行,但是函數(shù)的第一個參數(shù)還不是很明白。留待以后。
總結(jié)
future的設(shè)計理念很棒,在線程池/進程池和攜程中都存在future對象,是異步編程的核心。
ThreadPoolExecutor 讓線程的使用更加方便,減小了線程創(chuàng)建/銷毀的資源損耗,無需考慮線程間的復(fù)雜同步,方便主線程與子線程的交互。
線程池的抽象程度很高,多線程和多進程的編碼接口一致。
以上為個人經(jīng)驗,希望能給大家一個參考,也希望大家多多支持腳本之家。
相關(guān)文章
Linux上使用Python統(tǒng)計每天的鍵盤輸入次數(shù)
這篇文章主要介紹了Linux上使用Python統(tǒng)計每天的鍵盤輸入次數(shù),非常不錯,具有一定的參考借鑒價值,需要的朋友可以參考下2019-04-04詳解pandas刪除缺失數(shù)據(jù)(pd.dropna()方法)
這篇文章主要介紹了pandas刪除缺失數(shù)據(jù)(pd.dropna()方法),文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-06-06pytorch 如何把圖像數(shù)據(jù)集進行劃分成train,test和val
這篇文章主要介紹了pytorch 把圖像數(shù)據(jù)集進行劃分成train,test和val的操作,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-05-05解決Pytorch在測試與訓(xùn)練過程中的驗證結(jié)果不一致問題
這篇文章主要介紹了解決Pytorch在測試與訓(xùn)練過程中的驗證結(jié)果不一致問題,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-06-06