PythonThreadPoolExecutor線程池問題怎麼解決

概念

Python中已經有了threading模組，為什麼還需要執行緒池呢，線程池又是什麼東西呢？

以爬蟲為例，需要控制同時爬取的線程數，例子中創建了20個線程，而同時只允許3個線程在運行，但是20個線程都需要創建和銷毀，線程的創建是需要消耗系統資源的，有沒有更好的方案呢？

其實只需要三個執行緒就行了，每個執行緒各分配一個任務，剩下的任務排隊等待，當某個執行緒完成了任務的時候，排隊任務就可以安排給這個執行緒繼續執行。

這就是線程池的想法（當然沒這麼簡單），但是自己寫線程池很難寫的比較完美，還需要考慮複雜情況下的線程同步，很容易發生死鎖。

從Python3.2開始，標準函式庫為我們提供了concurrent.futures模組，它提供了ThreadPoolExecutor和ProcessPoolExecutor 兩個類，實現了對threading和multiprocessing的進一步抽象（這裡主要關注線程池），不僅可以幫我們自動調度線程，還可以做到：

• 主執行緒可以取得某一個執行緒（或任務的）的狀態，以及回傳值。

• 當一個執行緒完成的時候，主執行緒能夠立即知道。

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

實例

簡單使用

from concurrent.futures import ThreadPoolExecutor
import time
 
# 参数times用来模拟网络请求的时间
def get_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就完成了。

使用cancel()方法可以取消提交的任務，如果任務已經在執行緒池中運行了，就取消不了。這個例子中，線程池的大小設定為2，任務已經在運行了，所以取消失敗。如果改變線程池的大小為1，那麼先提交的是task1，task2還在排隊等候，這是時候就可以成功取消。

使用result()方法可以取得任務的回傳值。查看內部程式碼，發現這個方法是阻塞的。

as_completed

上面雖然提供了判斷任務是否結束的方法，但是不能在主執行緒一直判斷啊。

有時候我們是得知某個任務結束了，就去取得結果，而不是一直判斷每個任務有沒有結束。

這是就可以使用as_completed方法一次取出所有任務的結果。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
# 参数times用来模拟网络请求的时间
def get_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用来模拟网络请求的时间
def get_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用来模拟网络请求的时间
def get_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，表示第一個任務完成就停止等待。

原始碼分析

cocurrent.future模組中的future的意思是未來對象，可以把它理解為一個在未來完成的操作，這是非同步程式設計的基礎。

在執行緒池submit()之後，回傳的就是這個future對象，回傳的時候任務並沒有完成，但會在將來完成。

也可以稱為task的回傳容器，這個裡面會儲存task的結果和狀態。

那ThreadPoolExecutor內部是如何操作這個物件的呢？

下面簡單介紹ThreadPoolExecutor的部分程式碼：

1.init方法

init方法中主要重要的就是任務隊列和線程集合，在其他方法中需要使用到。

2.submit方法

submit中有兩個重要的對象，_base.Future()和_WorkItem()對象，_WorkItem()對象負責執行任務和對future對象進行設置，最後會將future對象返回，可以看到整個過程是立即返回的，沒有阻塞。

3.adjust_thread_count方法

這個方法的意義很好理解，主要是建立指定的執行緒數。但實作上有點難以理解，例如執行緒執行函數中的weakref.ref，涉及到了弱引用等概念，留待以後理解。

4._WorkItem物件

#_WorkItem物件的職責就是執行任務和設定結果。這裡面主要複雜的還是self.future.set_result(result)。

5.執行緒執行函數--_worker

#這是執行緒池建立執行緒時指定的函數入口，主要是從佇列中依次取出task執行，但是函數的第一個參數還不是很明白。留待以後。

以上是PythonThreadPoolExecutor線程池問題怎麼解決的詳細內容。更多資訊請關注PHP中文網其他相關文章！