執行緒池 - tw511教學網

假設我們必須多執行緒任務建立大量執行緒。由於執行緒太多，因此可能會有很多效能問題，這在計算上會是最昂貴的。一個主要問題可能是吞吐量受限。我們可以通過建立一個執行緒池來解決這個問題。一個執行緒池可以被定義為一組預先範例化和空閒的執行緒，它們隨時可以開始工作。建立執行緒池比我們需要執行大量任務時為每個任務範例化新執行緒更受歡迎。執行緒池可以管理大量執行緒的並行執行，如下所示 -

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

Python模組 - Concurrent.futures

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

在後面的章節中，我們將學習concurrent.futures模組中的類。

執行者類

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

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

在上面的例子中，一個ThreadPoolExecutor已經由5個執行緒構造而成。然後，在提供訊息之前等待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/',
   'https://www.tw511.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
'https://www.tw511.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指令碼的輸出 -