Python之ThreadPoolExecutor線程池問(wèn)題

更新時(shí)間：2023年03月14日 10:16:22 作者：程序猿-張益達(dá)

這篇文章主要介紹了Python之ThreadPoolExecutor線程池問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

概念

Python中已經(jīng)有了threading模塊，為什么還需要線程池呢，線程池又是什么東西呢？

以爬蟲(chóng)為例，需要控制同時(shí)爬取的線程數(shù)，例子中創(chuàng)建了20個(gè)線程，而同時(shí)只允許3個(gè)線程在運(yùn)行，但是20個(gè)線程都需要?jiǎng)?chuàng)建和銷(xiāo)毀，線程的創(chuàng)建是需要消耗系統(tǒng)資源的，有沒(méi)有更好的方案呢？

其實(shí)只需要三個(gè)線程就行了，每個(gè)線程各分配一個(gè)任務(wù)，剩下的任務(wù)排隊(duì)等待，當(dāng)某個(gè)線程完成了任務(wù)的時(shí)候，排隊(duì)任務(wù)就可以安排給這個(gè)線程繼續(xù)執(zhí)行。

這就是線程池的思想（當(dāng)然沒(méi)這么簡(jiǎn)單），但是自己編寫(xiě)線程池很難寫(xiě)的比較完美，還需要考慮復(fù)雜情況下的線程同步，很容易發(fā)生死鎖。

從Python3.2開(kāi)始，標(biāo)準(zhǔn)庫(kù)為我們提供了concurrent.futures模塊，它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個(gè)類(lèi)，實(shí)現(xiàn)了對(duì)threading和multiprocessing的進(jìn)一步抽象（這里主要關(guān)注線程池），不僅可以幫我們自動(dòng)調(diào)度線程，還可以做到：

主線程可以獲取某一個(gè)線程（或者任務(wù)的）的狀態(tài)，以及返回值。
當(dāng)一個(gè)線程完成的時(shí)候，主線程能夠立即知道。
讓多線程和多進(jìn)程的編碼接口一致。

實(shí)例

簡(jiǎn)單使用

from concurrent.futures import ThreadPoolExecutor
import time
 
# 參數(shù)times用來(lái)模擬網(wǎng)絡(luò)請(qǐng)求的時(shí)間
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
# 通過(guò)submit函數(shù)提交執(zhí)行的函數(shù)到線程池中，submit函數(shù)立即返回，不阻塞
task1 = executor.submit(get_html, (3))
task2 = executor.submit(get_html, (2))
# done方法用于判定某個(gè)任務(wù)是否完成
print(task1.done())
# cancel方法用于取消某個(gè)任務(wù),該任務(wù)沒(méi)有放入線程池中才能取消成功
print(task2.cancel())
time.sleep(4)
print(task1.done())
# result方法可以獲取task的執(zhí)行結(jié)果
print(task1.result())
 
# 執(zhí)行結(jié)果
# False  # 表明task1未執(zhí)行完成
# False  # 表明task2取消失敗，因?yàn)橐呀?jīng)放入了線程池中
# get page 2s finished
# get page 3s finished
# True  # 由于在get page 3s finished之后才打印，所以此時(shí)task1必然完成了
# 3     # 得到task1的任務(wù)返回值

ThreadPoolExecutor構(gòu)造實(shí)例的時(shí)候，傳入max_workers參數(shù)來(lái)設(shè)置線程池中最多能同時(shí)運(yùn)行的線程數(shù)目。

使用submit函數(shù)來(lái)提交線程需要執(zhí)行的任務(wù)（函數(shù)名和參數(shù)）到線程池中，并返回該任務(wù)的句柄（類(lèi)似于文件、畫(huà)圖），注意submit()不是阻塞的，而是立即返回。

通過(guò)submit函數(shù)返回的任務(wù)句柄，能夠使用done()方法判斷該任務(wù)是否結(jié)束。上面的例子可以看出，由于任務(wù)有2s的延時(shí)，在task1提交后立刻判斷，task1還未完成，而在延時(shí)4s之后判斷，task1就完成了。

使用cancel()方法可以取消提交的任務(wù)，如果任務(wù)已經(jīng)在線程池中運(yùn)行了，就取消不了。這個(gè)例子中，線程池的大小設(shè)置為2，任務(wù)已經(jīng)在運(yùn)行了，所以取消失敗。如果改變線程池的大小為1，那么先提交的是task1，task2還在排隊(duì)等候，這是時(shí)候就可以成功取消。

使用result()方法可以獲取任務(wù)的返回值。查看內(nèi)部代碼，發(fā)現(xiàn)這個(gè)方法是阻塞的。

as_completed

上面雖然提供了判斷任務(wù)是否結(jié)束的方法，但是不能在主線程中一直判斷啊。

有時(shí)候我們是得知某個(gè)任務(wù)結(jié)束了，就去獲取結(jié)果，而不是一直判斷每個(gè)任務(wù)有沒(méi)有結(jié)束。

這是就可以使用as_completed方法一次取出所有任務(wù)的結(jié)果。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
# 參數(shù)times用來(lái)模擬網(wǎng)絡(luò)請(qǐng)求的時(shí)間
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()方法是一個(gè)生成器，在沒(méi)有任務(wù)完成的時(shí)候，會(huì)阻塞，在有某個(gè)任務(wù)完成的時(shí)候，會(huì)yield這個(gè)任務(wù)，就能執(zhí)行for循環(huán)下面的語(yǔ)句，然后繼續(xù)阻塞住，循環(huán)到所有的任務(wù)結(jié)束。

從結(jié)果也可以看出，先完成的任務(wù)會(huì)先通知主線程。

map

除了上面的as_completed方法，還可以使用executor.map方法，但是有一點(diǎn)不同。

from concurrent.futures import ThreadPoolExecutor
import time
 
# 參數(shù)times用來(lái)模擬網(wǎng)絡(luò)請(qǐng)求的時(shí)間
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方法，無(wú)需提前使用submit方法，map方法與python標(biāo)準(zhǔn)庫(kù)中的map含義相同，都是將序列中的每個(gè)元素都執(zhí)行同一個(gè)函數(shù)。

上面的代碼就是對(duì)urls的每個(gè)元素都執(zhí)行get_html函數(shù)，并分配各線程池?？梢钥吹綀?zhí)行結(jié)果與上面的as_completed方法的結(jié)果不同，輸出順序和urls列表的順序相同，就算2s的任務(wù)先執(zhí)行完成，也會(huì)先打印出3s的任務(wù)先完成，再打印2s的任務(wù)完成。

wait

wait方法可以讓主線程阻塞，直到滿(mǎn)足設(shè)定的要求。

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time
 
# 參數(shù)times用來(lái)模擬網(wǎng)絡(luò)請(qǐng)求的時(shí)間
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個(gè)參數(shù)，等待的任務(wù)序列、超時(shí)時(shí)間以及等待條件。

等待條件return_when默認(rèn)為ALL_COMPLETED，表明要等待所有的任務(wù)都結(jié)束。

可以看到運(yùn)行結(jié)果中，確實(shí)是所有任務(wù)都完成了，主線程才打印出main。

等待條件還可以設(shè)置為FIRST_COMPLETED，表示第一個(gè)任務(wù)完成就停止等待。

源碼分析

cocurrent.future模塊中的future的意思是未來(lái)對(duì)象，可以把它理解為一個(gè)在未來(lái)完成的操作，這是異步編程的基礎(chǔ) 。

在線程池submit()之后，返回的就是這個(gè)future對(duì)象，返回的時(shí)候任務(wù)并沒(méi)有完成，但會(huì)在將來(lái)完成。

也可以稱(chēng)之為task的返回容器，這個(gè)里面會(huì)存儲(chǔ)task的結(jié)果和狀態(tài)。

那ThreadPoolExecutor內(nèi)部是如何操作這個(gè)對(duì)象的呢？

下面簡(jiǎn)單介紹ThreadPoolExecutor的部分代碼：

1.init方法

init方法中主要重要的就是任務(wù)隊(duì)列和線程集合，在其他方法中需要使用到。

2.submit方法

submit中有兩個(gè)重要的對(duì)象，_base.Future()和_WorkItem()對(duì)象，_WorkItem()對(duì)象負(fù)責(zé)運(yùn)行任務(wù)和對(duì)future對(duì)象進(jìn)行設(shè)置，最后會(huì)將future對(duì)象返回，可以看到整個(gè)過(guò)程是立即返回的，沒(méi)有阻塞。