[python] ThreadPoolExecutor線程池


初識
Python中已經有了threading模塊,為什么還需要線程池呢,線程池又是什么東西呢?在介紹線程同步的信號量機制的時候,舉得例子是爬蟲的例子,需要控制同時爬取的線程數,例子中創建了20個線程,而同時只允許3個線程在運行,但是20個線程都需要創建和銷毀,線程的創建是需要消耗系統資源的,有沒有更好的方案呢?其實只需要三個線程就行了,每個線程各分配一個任務,剩下的任務排隊等待,當某個線程完成了任務的時候,排隊任務就可以安排給這個線程繼續執行。

這就是線程池的思想(當然沒這么簡單),但是自己編寫線程池很難寫的比較完美,還需要考慮復雜情況下的線程同步,很容易發生死鎖。從Python3.2開始,標准庫為我們提供了concurrent.futures模塊,它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個類,實現了對threading和multiprocessing的進一步抽象(這里主要關注線程池),不僅可以幫我們自動調度線程,還可以做到:

主線程可以獲取某一個線程(或者任務的)的狀態,以及返回值。
當一個線程完成的時候,主線程能夠立即知道。
讓多線程和多進程的編碼接口一致。
實例
簡單使用
from concurrent.futures import ThreadPoolExecutor
import time

參數times用來模擬網絡請求的時間

def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times

executor = ThreadPoolExecutor(max_workers=2)

通過submit函數提交執行的函數到線程池中,submit函數立即返回,不阻塞

task1 = executor.submit(get_html, (3))
task2 = executor.submit(get_html, (2))

done方法用於判定某個任務是否完成

print(task1.done())

cancel方法用於取消某個任務,該任務沒有放入線程池中才能取消成功

print(task2.cancel())
time.sleep(4)
print(task1.done())

result方法可以獲取task的執行結果

print(task1.result())

執行結果

False # 表明task1未執行完成

False # 表明task2取消失敗,因為已經放入了線程池中

get page 2s finished

get page 3s finished

True # 由於在get page 3s finished之后才打印,所以此時task1必然完成了

3 # 得到task1的任務返回值

ThreadPoolExecutor構造實例的時候,傳入max_workers參數來設置線程池中最多能同時運行的線程數目。
使用submit函數來提交線程需要執行的任務(函數名和參數)到線程池中,並返回該任務的句柄(類似於文件、畫圖),注意submit()不是阻塞的,而是立即返回。
通過submit函數返回的任務句柄,能夠使用done()方法判斷該任務是否結束。上面的例子可以看出,由於任務有2s的延時,在task1提交后立刻判斷,task1還未完成,而在延時4s之后判斷,task1就完成了。
使用cancel()方法可以取消提交的任務,如果任務已經在線程池中運行了,就取消不了。這個例子中,線程池的大小設置為2,任務已經在運行了,所以取消失敗。如果改變線程池的大小為1,那么先提交的是task1,task2還在排隊等候,這是時候就可以成功取消。
使用result()方法可以獲取任務的返回值。查看內部代碼,發現這個方法是阻塞的。
as_completed
上面雖然提供了判斷任務是否結束的方法,但是不能在主線程中一直判斷啊。有時候我們是得知某個任務結束了,就去獲取結果,而不是一直判斷每個任務有沒有結束。這是就可以使用as_completed方法一次取出所有任務的結果。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

參數times用來模擬網絡請求的時間

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))

執行結果

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()方法是一個生成器,在沒有任務完成的時候,會阻塞,在有某個任務完成的時候,會yield這個任務,就能執行for循環下面的語句,然后繼續阻塞住,循環到所有的任務結束。從結果也可以看出,先完成的任務會先通知主線程。

map
除了上面的as_completed方法,還可以使用executor.map方法,但是有一點不同。

from concurrent.futures import ThreadPoolExecutor
import time

參數times用來模擬網絡請求的時間

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))

執行結果

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標准庫中的map含義相同,都是將序列中的每個元素都執行同一個函數。上面的代碼就是對urls的每個元素都執行get_html函數,並分配各線程池。可以看到執行結果與上面的as_completed方法的結果不同,輸出順序和urls列表的順序相同,就算2s的任務先執行完成,也會先打印出3s的任務先完成,再打印2s的任務完成。

wait
wait方法可以讓主線程阻塞,直到滿足設定的要求。

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time

參數times用來模擬網絡請求的時間

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")

執行結果

get page 2s finished

get page 3s finished

get page 4s finished

main

wait方法接收3個參數,等待的任務序列、超時時間以及等待條件。等待條件return_when默認為ALL_COMPLETED,表明要等待所有的任務都結束。可以看到運行結果中,確實是所有任務都完成了,主線程才打印出main。等待條件還可以設置為FIRST_COMPLETED,表示第一個任務完成就停止等待。

源碼分析
cocurrent.future模塊中的future的意思是未來對象,可以把它理解為一個在未來完成的操作,這是異步編程的基礎 。在線程池submit()之后,返回的就是這個future對象,返回的時候任務並沒有完成,但會在將來完成。也可以稱之為task的返回容器,這個里面會存儲task的結果和狀態。那ThreadPoolExecutor內部是如何操作這個對象的呢?

下面簡單介紹ThreadPoolExecutor的部分代碼:

init方法

ThreadPoolExecutor_init方法.png

init方法中主要重要的就是任務隊列和線程集合,在其他方法中需要使用到。
submit方法

ThreadPoolExecutor_submit方法.png
submit中有兩個重要的對象,_base.Future()和_WorkItem()對象,_WorkItem()對象負責運行任務和對future對象進行設置,最后會將future對象返回,可以看到整個過程是立即返回的,沒有阻塞。

adjust_thread_count方法

ThreadPoolExecutor_adjust_thread_count方法.png

這個方法的含義很好理解,主要是創建指定的線程數。但是實現上有點難以理解,比如線程執行函數中的weakref.ref,涉及到了弱引用等概念,留待以后理解。

_WorkItem對象

_WorkItem類.png

_WorkItem對象的職責就是執行任務和設置結果。這里面主要復雜的還是self.future.set_result(result)。
線程執行函數--_worker

ThreadPoolExecutor_線程執行函數_worker.png

這是線程池創建線程時指定的函數入口,主要是從隊列中依次取出task執行,但是函數的第一個參數還不是很明白。留待以后。

總結
future的設計理念很棒,在線程池/進程池和攜程中都存在future對象,是異步編程的核心。
ThreadPoolExecutor 讓線程的使用更加方便,減小了線程創建/銷毀的資源損耗,無需考慮線程間的復雜同步,方便主線程與子線程的交互。
線程池的抽象程度很高,多線程和多進程的編碼接口一致。
未完成
對future模塊的理解。
weakref.ref是什么?
線程執行函數入口_worker的第一個參數的意思。
參考
Python並發編程之線程池/進程池
Python3高級編程和異步IO並發編程

作者:StormZhu
鏈接:https://www.jianshu.com/p/b9b3d66aa0be
來源:簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權,非商業轉載請注明出處。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM