jlzzjlzz亚洲乱熟在线播放

系統城裝機大師 - 唯一官網:www.farandoo.com!

當前位置:首頁 > 腳本中心 > python > 詳細頁面

python線程池 ThreadPoolExecutor 的用法示例

時間:2020-10-11來源:www.farandoo.com作者:電腦系統城

前言

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

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

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

線程池的基本使用

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 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

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

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

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

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

主要方法

  • wait

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

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

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

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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

1.代碼中返回的條件是:當完成第一個任務的時候,就停止等待,繼續主線程任務

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

  • as_completed

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

用法如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 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。

用法如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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 的任務先執行完成,也會先打印前面提交的任務返回的結果。

多線程實戰

以某網站為例,演示線程池和單線程兩種方式爬取的差異

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# coding: utf-8
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import json
from requests import adapters
 
from proxy import get_proxies
 
headers = {
 "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_data
 
def 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()

運行結果:

單線程實戰

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

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# coding: utf-8
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import json
from requests import adapters
 
from proxy import get_proxies
 
headers = {
 "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_data
 
def 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 秒。真是肉眼可見的差距??!如果數據量大的話,運行時間差距會更大!

以上就是python線程池 ThreadPoolExecutor 的用法示例的詳細內容,更多關于python線程池 ThreadPoolExecutor 的用法及實戰的資料請關注腳本之家其它相關文章!

分享到:

相關信息

系統教程欄目

欄目熱門教程

人氣教程排行

站長推薦

熱門系統下載