初識

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

這就是執行緒池的思想（當然沒這麼簡單），但是自己編寫執行緒池很難寫的比較完美，還需要考慮複雜情況下的執行緒同步，很容易發生死鎖。從Python3.2

1. 主執行緒可以獲取某一個執行緒（或者任務的）的狀態，以及返回值。
2. 當一個執行緒完成的時候，主執行緒能夠立即知道。
3. 讓多執行緒和多程序的編碼介面一致。

例項

簡單使用

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的任務返回值
 

1. ThreadPoolExecutor構造例項的時候，傳入max_workers引數來設定執行緒池中最多能同時執行的執行緒數目。
2. 使用submit函式來提交執行緒需要執行的任務（函式名和引數）到執行緒池中，並返回該任務的控制代碼（類似於檔案、畫圖），注意submit()不是阻塞的，而是立即返回。
3. 通過submit函式返回的任務控制代碼，能夠使用done()方法判斷該任務是否結束。上面的例子可以看出，由於任務有2s的延時，在task1提交後立刻判斷，task1還未完成，而在延時4s之後判斷，task1就完成了。
4. 使用cancel()方法可以取消提交的任務，如果任務已經線上程池中運行了，就取消不了。這個例子中，執行緒池的大小設定為2，任務已經在運行了，所以取消失敗。如果改變執行緒池的大小為1，那麼先提交的是task1
   ，task2還在排隊等候，這是時候就可以成功取消。
5. 使用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方法

     ThreadPoolExecutor_init方法.png
   init方法中主要重要的就是任務佇列和執行緒集合，在其他方法中需要使用到。

2. submit方法

     ThreadPoolExecutor_submit方法.png

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

3. adjust_thread_count方法

     ThreadPoolExecutor_adjust_thread_count方法.png
   

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

4. _WorkItem物件

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

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

     ThreadPoolExecutor_執行緒執行函式_worker.png
   

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

總結

• future的設計理念很棒，線上程池/程序池和攜程中都存在future物件，是非同步程式設計的核心。
• ThreadPoolExecutor 讓執行緒的使用更加方便，減小了執行緒建立/銷燬的資源損耗，無需考慮執行緒間的複雜同步，方便主執行緒與子執行緒的互動。
• 執行緒池的抽象程度很高，多執行緒和多程序的編碼介面一致。