python 中線程池的使用

python中已經有了threading模組，為什麼還需要執行緒池呢，執行緒池又是什麼東西呢？在介紹執行緒同步的訊號量機制的時候，舉得例子是爬蟲的例子，需要控制同時爬取的執行緒數，例子中建立了20個執行緒，而同時只允許3個執行緒在執行，但是20個執行緒都需要建立和銷毀，執行緒的建立是需要消耗系統資源的，有沒有更好的方案呢？其實只需要三個執行緒就行了，每個執行緒各分配乙個任務，剩下的任務排隊等待，當某個執行緒完成了任務的時候，排隊任務就可以安排給這個執行緒繼續執行。

這就是執行緒池的思想（當然沒這麼簡單），但是自己編寫執行緒池很難寫的比較完美，還需要考慮複雜情況下的執行緒同步，很容易發生死鎖。從python3.2開始，標準庫為我們提供了concurrent.futures模組，它提供了threadpoolexecutor和processpoolexecutor兩個類，實現了對threading和multiprocessing的進一步抽象（這裡主要關注執行緒池），不僅可以幫我們自動排程執行緒，還可以做到：

主線程可以獲取某乙個執行緒（或者任務的）的狀態，以及返回值。

當乙個執行緒完成的時候，主線程能夠立即知道。

讓多執行緒和多程序的編碼介面一致。

from concurrent.futures import threadpoolexecutor
import time
# 引數times用來模擬網路請求的時間
defget_html
(times)
:    time.sleep(times)
print
("get page {}s finished"
.format
(times)
)return times
executor = threadpoolexecutor(max_workers=2)
# 通過submit函式提交執行的函式到執行緒池中，submit函式立即返回，不阻塞
task1 = executor.submit(get_html,(3
))task2 = executor.submit(get_html,(2
))# done方法用於判定某個任務是否完成
print
(task1.done())
# cancel方法用於取消某個任務,該任務沒有放入執行緒池中才能取消成功
print
(task2.cancel())
time.sleep(4)
print
(task1.done())
# result方法可以獲取task的執行結果
print
(task1.result())
# 執行結果
# false  # 表明task1未執行完成
# false  # 表明task2取消失敗，因為已經放入了執行緒池中
# get page 2s finished
# get page 3s finished
# true  # 由於在get page 3s finished之後才列印，所以此時task1必然完成了
# 3     # 得到task1的任務返回值

threadpoolexecutor構造例項的時候，傳入max_workers引數來設定執行緒池中最多能同時執行的執行緒數目。

使用submit函式來提交執行緒需要執行的任務（函式名和引數）到執行緒池中，並返回該任務的控制代碼（類似於檔案、畫圖），注意submit()不是阻塞的，而是立即返回。

通過submit函式返回的任務控制代碼，能夠使用done()方法判斷該任務是否結束。上面的例子可以看出，由於任務有2s的延時，在task1提交後立刻判斷，task1還未完成，而在延時4s之後判斷，task1就完成了。

使用result()方法可以獲取任務的返回值。這個方法是阻塞的。

as_completed

上面雖然提供了判斷任務是否結束的方法，但是不能在主線程中一直判斷啊。有時候我們是得知某個任務結束了，就去獲取結果，而不是一直判斷每個任務有沒有結束。這是就可以使用as_completed方法一次取出所有任務的結果。

from concurrent.futures import threadpoolexecutor, as_completed
import time
# 引數times用來模擬網路請求的時間
defget_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)
)# 執行結果
# 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()方法是乙個生成器，在沒有任務完成的時候，會阻塞，在有某個任務完成的時候，會yield這個任務，就能執行for迴圈下面的語句，然後繼續阻塞住，迴圈到所有的任務結束。從結果也可以看出，先完成的任務會先通知主線程。

map

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

from concurrent.futures import threadpoolexecutor
import time
# 引數times用來模擬網路請求的時間
defget_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)
)# 執行結果
# 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標準庫中的map含義相同，都是將序列中的每個元素都執行同乙個函式。上面的**就是對urls的每個元素都執行get_html函式，並分配各執行緒池。可以看到執行結果與上面的as_completed方法的結果不同，輸出順序和urls列表的順序相同，就算2s的任務先執行完成，也會先列印出3s的任務先完成，再列印2s的任務完成。

wait

wait方法可以讓主線程阻塞，直到滿足設定的要求。

from concurrent.futures import threadpoolexecutor, wait, all_completed, first_completed
import time
# 引數times用來模擬網路請求的時間
defget_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"
)# 執行結果 
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main

wait方法接收3個引數，等待的任務序列、超時時間以及等待條件。等待條件return_when預設為all_completed，表明要等待所有的任務都結束。可以看到執行結果中，確實是所有任務都完成了，主線程才列印出main。等待條件還可以設定為first_completed，表示第乙個任務完成就停止等待。

python 中線程池的使用

Python中線程池的實現

Java中線程池的使用

c 中線程池

python 中線程池的使用

Python中線程池的實現

Java中線程池的使用

c 中線程池

相關推薦