Python中的併發——執行緒池

假設我們需要為多執行緒任務建立大量執行緒。由於執行緒過多可能導致許多效能問題，這在計算上將非常昂貴，主要問題可能是吞吐量受限。我們可以透過建立執行緒池來解決這個問題。執行緒池可以定義為一組預例項化且處於空閒狀態的執行緒，它們隨時準備接收工作。當我們需要執行大量任務時，建立執行緒池優於為每個任務例項化新執行緒。執行緒池可以如下管理大量執行緒的併發執行：

如果執行緒池中的一個執行緒完成執行，則可以重用該執行緒。
如果一個執行緒終止，將建立另一個執行緒來替換該執行緒。

Python模組 – concurrent.futures

Python標準庫包含**concurrent.futures**模組。此模組新增到Python 3.2中，為開發人員提供了一個高階介面來啟動非同步任務。它是Python的執行緒和多程序模組頂層的一個抽象層，用於提供使用執行緒池或程序池執行任務的介面。

在接下來的章節中，我們將學習concurrent.futures模組的不同類。

Executor類

**Executor**是**concurrent.futures** Python模組的抽象類。它不能直接使用，我們需要使用以下具體子類之一：

ThreadPoolExecutor
ProcessPoolExecutor

ThreadPoolExecutor – 一個具體子類

它是Executor類的具體子類之一。該子類使用多執行緒，我們得到一個用於提交任務的執行緒池。該池將任務分配給可用的執行緒並安排它們執行。

如何建立一個ThreadPoolExecutor？

藉助**concurrent.futures**模組及其具體子類**Executor**，我們可以輕鬆建立執行緒池。為此，我們需要構造一個**ThreadPoolExecutor**，其中包含我們想要在池中使用的執行緒數。預設情況下，數量為5。然後我們可以將任務提交到執行緒池。當我們**submit()**一個任務時，我們會得到一個**Future**。Future物件有一個名為**done()**的方法，它告訴我們future是否已解析。有了這個，該特定future物件的值已被設定。當任務完成時，執行緒池執行器將值設定為future物件。

示例

from concurrent.futures import ThreadPoolExecutor
from time import sleep
def task(message):
   sleep(2)
   return message

def main():
   executor = ThreadPoolExecutor(5)
   future = executor.submit(task, ("Completed"))
   print(future.done())
   sleep(2)
   print(future.done())
   print(future.result())
if __name__ == '__main__':
main()

輸出

False
True
Completed

在上面的示例中，已構造了一個具有5個執行緒的**ThreadPoolExecutor**。然後將一個任務提交到執行緒池執行器，該任務將在發出訊息之前等待2秒。從輸出中可以看出，任務直到2秒後才完成，因此對**done()**的第一次呼叫將返回False。2秒後，任務完成，我們透過在其上呼叫**result()**方法來獲取future的結果。

例項化ThreadPoolExecutor – 上下文管理器

例項化**ThreadPoolExecutor**的另一種方法是藉助上下文管理器。它的工作方式類似於上述示例中使用的方法。使用上下文管理器的主要優點是它在語法上看起來很好。可以使用以下程式碼進行例項化：

with ThreadPoolExecutor(max_workers = 5) as executor

示例

下面的示例取自Python文件。在這個例子中，首先必須匯入**concurrent.futures**模組。然後建立一個名為**load_url()**的函式，該函式將載入請求的url。然後，該函式建立具有池中5個執行緒的**ThreadPoolExecutor**。**ThreadPoolExecutor**已被用作上下文管理器。我們可以透過在其上呼叫**result()**方法來獲取future的結果。

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
   'http://www.cnn.com/',
   'http://europe.wsj.com/',
   'http://www.bbc.co.uk/',
   'http://some-made-up-domain.com/']

def load_url(url, timeout):
   with urllib.request.urlopen(url, timeout = timeout) as conn:
   return conn.read()

with concurrent.futures.ThreadPoolExecutor(max_workers = 5) as executor:

   future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
   for future in concurrent.futures.as_completed(future_to_url):
   url = future_to_url[future]
   try:
      data = future.result()
   except Exception as exc:
      print('%r generated an exception: %s' % (url, exc))
   else:
      print('%r page is %d bytes' % (url, len(data)))

輸出

以下是上述Python指令碼的輸出：

'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229313 bytes
'http://www.cnn.com/' page is 168933 bytes
'http://www.bbc.co.uk/' page is 283893 bytes
'http://europe.wsj.com/' page is 938109 bytes

Executor.map()函式的使用

Python **map()**函式廣泛用於許多工中。一項這樣的任務是將某個函式應用於可迭代物件中的每個元素。類似地，我們可以將迭代器的所有元素對映到一個函式，並將這些元素作為獨立作業提交到我們的**ThreadPoolExecutor**。考慮以下Python指令碼示例，以瞭解該函式的工作方式。

示例

在下面的示例中，map函式用於將**square()**函式應用於values陣列中的每個值。

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
   return n * n
def main():
   with ThreadPoolExecutor(max_workers = 3) as executor:
      results = executor.map(square, values)
for result in results:
      print(result)
if __name__ == '__main__':
   main()

輸出

上述Python指令碼生成以下輸出：

列印頁面