溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本文小編為大家詳細介紹“Python之ThreadPoolExecutor線程池問題怎么解決”，內容詳細，步驟清晰，細節處理妥當，希望這篇“Python之ThreadPoolExecutor線程池問題怎么解決”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來學習新知識吧。
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方法一次取出所有任務的結果。
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 successas_completed()方法是一個生成器，在沒有任務完成的時候，會阻塞，在有某個任務完成的時候，會yield這個任務，就能執行for循環下面的語句，然后繼續阻塞住，循環到所有的任務結束。
從結果也可以看出，先完成的任務會先通知主線程。
除了上面的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方法可以讓主線程阻塞，直到滿足設定的要求。
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
# mainwait方法接收3個參數，等待的任務序列、超時時間以及等待條件。
等待條件return_when默認為ALL_COMPLETED，表明要等待所有的任務都結束。
可以看到運行結果中，確實是所有任務都完成了，主線程才打印出main。
等待條件還可以設置為FIRST_COMPLETED，表示第一個任務完成就停止等待。
cocurrent.future模塊中的future的意思是未來對象，可以把它理解為一個在未來完成的操作，這是異步編程的基礎 。
在線程池submit()之后，返回的就是這個future對象，返回的時候任務并沒有完成，但會在將來完成。
也可以稱之為task的返回容器，這個里面會存儲task的結果和狀態。
那ThreadPoolExecutor內部是如何操作這個對象的呢？
下面簡單介紹ThreadPoolExecutor的部分代碼：
init方法中主要重要的就是任務隊列和線程集合，在其他方法中需要使用到。
submit中有兩個重要的對象，_base.Future()和_WorkItem()對象，_WorkItem()對象負責運行任務和對future對象進行設置，最后會將future對象返回，可以看到整個過程是立即返回的，沒有阻塞。
這個方法的含義很好理解，主要是創建指定的線程數。但是實現上有點難以理解，比如線程執行函數中的weakref.ref，涉及到了弱引用等概念，留待以后理解。
_WorkItem對象的職責就是執行任務和設置結果。這里面主要復雜的還是self.future.set_result(result)。
這是線程池創建線程時指定的函數入口，主要是從隊列中依次取出task執行，但是函數的第一個參數還不是很明白。留待以后。
讀到這里，這篇“Python之ThreadPoolExecutor線程池問題怎么解決”文章已經介紹完畢，想要掌握這篇文章的知識點還需要大家自己動手實踐使用過才能領會，如果想了解更多相關內容的文章，歡迎關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
