python執行緒池 ThreadPoolExecutor 的用法
✨ 前言
從Python3.2開始,標準庫為我們提供了 concurrent.futures 模組,它提供了 ThreadPoolExecutor (執行緒池)和ProcessPoolExecutor (程序池)兩個類。
相比 threading 等模組,該模組通過 submit 返回的是一個 future 物件,它是一個未來可期的物件,通過它可以獲悉執行緒的狀態主執行緒(或程序)中可以獲取某一個執行緒(程序)執行的狀態或者某一個任務執行的狀態及返回值:
- 主執行緒可以獲取某一個執行緒(或者任務的)的狀態,以及返回值。
- 當一個執行緒完成的時候,主執行緒能夠立即知道。
- 讓多執行緒和多程序的編碼介面一致。
✨ 執行緒池的基本使用
# coding: utf-8 from concurrent.futures import ThreadPoolExecutor import time def spider(page): time.sleep(page) print(f"crawl task{page} finished") return page with ThreadPoolExecutor(max_workers=5) as t: # 建立一個最大容納數量為5的執行緒池 task1 = t.submit(spider, 1) task2= t.submit(spider, 2) # 通過submit提交執行的函式到執行緒池中 task3 = t.submit(spider, 3) print(f"task1: {task1.done()}") # 通過done來判斷執行緒是否完成 print(f"task2: {task2.done()}") print(f"task3: {task3.done()}") time.sleep(2.5) print(f"task1: {task1.done()}") print(f"task2: {task2.done()}")print(f"task3: {task3.done()}") print(task1.result()) # 通過result來獲取返回值
執行結果如下:
task1: False task2: False task3: False crawl task1 finished crawl task2 finished task1: True task2: True task3: False 1 crawl task3 finished
-
使用 with 語句 ,通過 ThreadPoolExecutor 構造例項,同時傳入 max_workers 引數來設定執行緒池中最多能同時執行的執行緒數目。
-
使用 submit 函式來提交執行緒需要執行的任務到執行緒池中,並返回該任務的控制代碼(類似於檔案、畫圖),注意 submit() 不是阻塞的,而是立即返回。
-
通過使用 done() 方法判斷該任務是否結束。上面的例子可以看出,提交任務後立即判斷任務狀態,顯示四個任務都未完成。在延時2.5後,task1 和 task2 執行完畢,task3 仍在執行中。
-
使用 result() 方法可以獲取任務的返回值。
✨ 主要方法:
wait
wait(fs, timeout=None, return_when=ALL_COMPLETED)
wait 接受三個引數:
fs: 表示需要執行的序列
timeout: 等待的最大時間,如果超過這個時間即使執行緒未執行完成也將返回
return_when:表示wait返回結果的條件,預設為 ALL_COMPLETED 全部執行完成再返回
還是用上面那個例子來熟悉用法
示例:
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED, ALL_COMPLETED import time def spider(page): time.sleep(page) print(f"crawl task{page} finished") return page with ThreadPoolExecutor(max_workers=5) as t: all_task = [t.submit(spider, page) for page in range(1, 5)] wait(all_task, return_when=FIRST_COMPLETED) print('finished') print(wait(all_task, timeout=2.5)) # 執行結果 crawl task1 finished finished crawl task2 finished crawl task3 finished DoneAndNotDoneFutures(done={<Future at 0x28c8710 state=finished returned int>, <Future at 0x2c2bfd0 state=finished returned int>, <Future at 0x2c1b7f0 state=finished returned int>}, not_done={<Future at 0x2c3a240 state=running>}) crawl task4 finished
- 程式碼中返回的條件是:當完成第一個任務的時候,就停止等待,繼續主執行緒任務
- 由於設定了延時, 可以看到最後只有 task4 還在執行中
as_completed
上面雖然提供了判斷任務是否結束的方法,但是不能在主執行緒中一直判斷啊。最好的方法是當某個任務結束了,就給主執行緒返回結果,而不是一直判斷每個任務是否結束。
ThreadPoolExecutorThreadPoolExecutor 中 的 as_completed() 就是這樣一個方法,當子執行緒中的任務執行完後,直接用 result() 獲取返回結果
用法如下:
# coding: utf-8 from concurrent.futures import ThreadPoolExecutor, as_completed import time def spider(page): time.sleep(page) print(f"crawl task{page} finished") return page def main(): with ThreadPoolExecutor(max_workers=5) as t: obj_list = [] for page in range(1, 5): obj = t.submit(spider, page) obj_list.append(obj) for future in as_completed(obj_list): data = future.result() print(f"main: {data}") # 執行結果 crawl task1 finished main: 1 crawl task2 finished main: 2 crawl task3 finished main: 3 crawl task4 finished main: 4
as_completed() 方法是一個生成器,在沒有任務完成的時候,會一直阻塞,除非設定了 timeout。
當有某個任務完成的時候,會 yield 這個任務,就能執行 for 迴圈下面的語句,然後繼續阻塞住,迴圈到所有的任務結束。同時,先完成的任務會先返回給主執行緒
map
map(fn, *iterables, timeout=None)
fn: 第一個引數 fn 是需要執行緒執行的函式;
iterables:第二個引數接受一個可迭代物件;
timeout: 第三個引數 timeout 跟 wait() 的 timeout 一樣,但由於 map 是返回執行緒執行的結果,如果 timeout小於執行緒執行時間會拋異常 TimeoutError。
用法如下:
import time from concurrent.futures import ThreadPoolExecutor def spider(page): time.sleep(page) return page start = time.time() executor = ThreadPoolExecutor(max_workers=4) i = 1 for result in executor.map(spider, [2, 3, 1, 4]): print("task{}:{}".format(i, result)) i += 1 # 執行結果 task1:2 task2:3 task3:1 task4:4
使用 map 方法,無需提前使用 submit 方法,map 方法與 python 高階函式 map 的含義相同,都是將序列中的每個元素都執行同一個函式。
上面的程式碼對列表中的每個元素都執行 spider() 函式,並分配各執行緒池。
可以看到執行結果與上面的 as_completed() 方法的結果不同,輸出順序和列表的順序相同,就算 1s 的任務先執行完成,也會先列印前面提交的任務返回的結果。
原文:https://www.jianshu.com/p/6d6e4f745c27