初識

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的進(jìn)一步抽象（這里主要關(guān)注線程池），不僅可以幫我們自動調(diào)度線程，還可以做到：

主線程可以獲取某一個線程（或者任務(wù)的）的狀態(tài)，以及返回值。

當(dāng)一個線程完成的時候，主線程能夠立即知道。

讓多線程和多進(jìn)程的編碼接口一致。

實例

簡單使用

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 timesexecutor = 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 timesexecutor = 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 timesexecutor = ThreadPoolExecutor(max_workers=2) urls = [3, 2, 4] # 并不是真的urlfor 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í)行g(shù)et_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 timesexecutor = 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ù)完成就停止等待。

項目實例

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 timestask_pool = ThreadPoolExecutor(10) request_list=[] for next_ip in range (ip_start,ip_stop,256): if (next_ip+256)>ip_stop:breakrequest_list.append(task_pool.submit(self.Thread_ips,next_ip)) wait(request_list, return_when=ALL_COMPLETED)

總結(jié)

• future的設(shè)計理念很棒，在線程池/進(jìn)程池和攜程中都存在future對象，是異步編程的核心。
• ThreadPoolExecutor 讓線程的使用更加方便，減小了線程創(chuàng)建/銷毀的資源損耗，無需考慮線程間的復(fù)雜同步，方便主線程與子線程的交互。
• 線程池的抽象程度很高，多線程和多進(jìn)程的編碼接口一致。

?

?

?

?

?

?

總結(jié)

以上是生活随笔為你收集整理的python ThreadPoolExecutor线程池（实例）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。