前言

從Python3.2開始，標準庫為我們提供了 concurrent.futures 模塊，它提供了 ThreadPoolExecutor (線程池)和ProcessPoolExecutor (進程池)兩個類。

相比 threading 等模塊，該模塊通過 submit 返回的是一個 future 對象，它是一個未來可期的對象，通過它可以獲悉線程的狀態(tài)主線程(或進程)中可以獲取某一個線程(進程)執(zhí)行的狀態(tài)或者某一個任務執(zhí)行的狀態(tài)及返回值：

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

線程池的基本使用

# coding: utf-8from concurrent.futures import ThreadPoolExecutorimport timedef spider(page): time.sleep(page) print(f'crawl task{page} finished') return pagewith ThreadPoolExecutor(max_workers=5) as t: # 創(chuàng)建一個最大容納數(shù)量為5的線程池 task1 = t.submit(spider, 1) task2 = t.submit(spider, 2) # 通過submit提交執(zhí)行的函數(shù)到線程池中 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來獲取返回值

執(zhí)行結果如下:

task1: Falsetask2: Falsetask3: Falsecrawl task1 finishedcrawl task2 finishedtask1: Truetask2: Truetask3: False1crawl task3 finished

1.使用 with 語句 ，通過 ThreadPoolExecutor 構造實例，同時傳入 max_workers 參數(shù)來設置線程池中最多能同時運行的線程數(shù)目。

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

3.通過使用 done() 方法判斷該任務是否結束。上面的例子可以看出，提交任務后立即判斷任務狀態(tài)，顯示四個任務都未完成。在延時2.5后，task1 和 task2 執(zhí)行完畢，task3 仍在執(zhí)行中。

4.使用 result() 方法可以獲取任務的返回值。

主要方法

wait(fs, timeout=None, return_when=ALL_COMPLETED)

wait 接受三個參數(shù)：fs: 表示需要執(zhí)行的序列timeout: 等待的最大時間，如果超過這個時間即使線程未執(zhí)行完成也將返回return_when：表示wait返回結果的條件，默認為 ALL_COMPLETED 全部執(zhí)行完成再返回

還是用上面那個例子來熟悉用法示例：

from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED, ALL_COMPLETEDimport timedef spider(page): time.sleep(page) print(f'crawl task{page} finished') return pagewith 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 finishedfinishedcrawl task2 finishedcrawl task3 finishedDoneAndNotDoneFutures(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

1.代碼中返回的條件是：當完成第一個任務的時候，就停止等待，繼續(xù)主線程任務

2.由于設置了延時， 可以看到最后只有 task4 還在運行中

上面雖然提供了判斷任務是否結束的方法，但是不能在主線程中一直判斷啊。最好的方法是當某個任務結束了，就給主線程返回結果，而不是一直判斷每個任務是否結束。ThreadPoolExecutorThreadPoolExecutor 中 的 as_completed() 就是這樣一個方法，當子線程中的任務執(zhí)行完后，直接用 result() 獲取返回結果

用法如下：

# coding: utf-8from concurrent.futures import ThreadPoolExecutor, as_completedimport timedef spider(page): time.sleep(page) print(f'crawl task{page} finished') return pagedef 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}')# 執(zhí)行結果crawl task1 finishedmain: 1crawl task2 finishedmain: 2crawl task3 finishedmain: 3crawl task4 finishedmain: 4

as_completed() 方法是一個生成器，在沒有任務完成的時候，會一直阻塞，除非設置了 timeout。

當有某個任務完成的時候，會 yield 這個任務，就能執(zhí)行 for 循環(huán)下面的語句，然后繼續(xù)阻塞住，循環(huán)到所有的任務結束。同時，先完成的任務會先返回給主線程。

map(fn, *iterables, timeout=None)

fn： 第一個參數(shù) fn 是需要線程執(zhí)行的函數(shù)；iterables：第二個參數(shù)接受一個可迭代對象；timeout： 第三個參數(shù) timeout 跟 wait() 的 timeout 一樣，但由于 map 是返回線程執(zhí)行的結果，如果 timeout小于線程執(zhí)行時間會拋異常 TimeoutError。

用法如下:

import timefrom concurrent.futures import ThreadPoolExecutordef spider(page): time.sleep(page) return pagestart = time.time()executor = ThreadPoolExecutor(max_workers=4)i = 1for result in executor.map(spider, [2, 3, 1, 4]): print('task{}:{}'.format(i, result)) i += 1# 運行結果task1:2task2:3task3:1task4:4

使用 map 方法，無需提前使用 submit 方法，map 方法與 python 高階函數(shù) map 的含義相同，都是將序列中的每個元素都執(zhí)行同一個函數(shù)。

上面的代碼對列表中的每個元素都執(zhí)行 spider() 函數(shù)，并分配各線程池。

可以看到執(zhí)行結果與上面的 as_completed() 方法的結果不同，輸出順序和列表的順序相同，就算 1s 的任務先執(zhí)行完成，也會先打印前面提交的任務返回的結果。

多線程實戰(zhàn)

以某網(wǎng)站為例，演示線程池和單線程兩種方式爬取的差異

# coding: utf-8import requestsfrom concurrent.futures import ThreadPoolExecutor, as_completedimport timeimport jsonfrom requests import adaptersfrom proxy import get_proxiesheaders = { 'Host': 'splcgk.court.gov.cn', 'Origin': 'https://splcgk.court.gov.cn', 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36', 'Referer': 'https://splcgk.court.gov.cn/gzfwww/ktgg',}url = 'https://splcgk.court.gov.cn/gzfwww/ktgglist?pageNo=1'def spider(page): data = { 'bt': '', 'fydw': '', 'pageNum': page, } for _ in range(5): try: response = requests.post(url, headers=headers, data=data, proxies=get_proxies()) json_data = response.json() except (json.JSONDecodeError, adapters.SSLError): continue else: break else: return {} return json_datadef main(): with ThreadPoolExecutor(max_workers=10) as t: obj_list = [] begin = time.time() for page in range(1, 15): obj = t.submit(spider, page) obj_list.append(obj) for future in as_completed(obj_list): data = future.result() print(data) print(’*’ * 50) times = time.time() - begin print(times)if __name__ == '__main__': main()

運行結果：

單線程實戰(zhàn)

下面我們可以使用單線程來爬取，代碼基本和上面的一樣，加個單線程函數(shù)代碼如下:

# coding: utf-8import requestsfrom concurrent.futures import ThreadPoolExecutor, as_completedimport timeimport jsonfrom requests import adaptersfrom proxy import get_proxiesheaders = { 'Host': 'splcgk.court.gov.cn', 'Origin': 'https://splcgk.court.gov.cn', 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36', 'Referer': 'https://splcgk.court.gov.cn/gzfwww/ktgg',}url = 'https://splcgk.court.gov.cn/gzfwww/ktgglist?pageNo=1'def spider(page): data = { 'bt': '', 'fydw': '', 'pageNum': page, } for _ in range(5): try: response = requests.post(url, headers=headers, data=data, proxies=get_proxies()) json_data = response.json() except (json.JSONDecodeError, adapters.SSLError): continue else: break else: return {} return json_datadef single(): begin = time.time() for page in range(1, 15): data = spider(page) print(data) print(’*’ * 50) times = time.time() - begin print(times)if __name__ == '__main__': single()

運行結果:

可以看到，總共花了 19 秒。真是肉眼可見的差距啊！如果數(shù)據(jù)量大的話，運行時間差距會更大！

以上就是python線程池 ThreadPoolExecutor 的用法示例的詳細內容，更多關于python線程池 ThreadPoolExecutor 的用法及實戰(zhàn)的資料請關注好吧啦網(wǎng)其它相關文章！