進程池

進程池可以像創建和使用線程池一樣創建和使用。進程池可以被定義爲一組預先實例化和空閒的進程，它們隨時可以開始工作。當我們需要執行大量任務時，創建進程池優於爲每個任務實例化新進程。

Python模塊 - Concurrent.futures

Python標準庫有一個叫做concurrent.futures的模塊。這個模塊是在Python 3.2中添加的，爲開發人員提供了啓動異步任務的高級接口。它是Python的線程和多處理模塊的頂層的一個抽象層，用於提供使用線程或進程池運行任務的接口。

在後面的章節中，我們將要學習concurrent.futures模塊的不同子類。

執行者類

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

ThreadPoolExecutor
ProcessPoolExecutor

ProcessPoolExecutor - 一個具體的子類

它是Executor類的具體子類之一。它使用多重處理，並且我們獲得提交任務的過程池。此池將任務分配給可用的進程並安排它們運行。

如何創建一個ProcessPoolExecutor？
通過concurrent.futures模塊及其具體子類Executor的幫助，可以輕鬆創建一個過程池。爲此，需要構建一個ProcessPoolExecutor，其中包含需要的池中的進程數。默認情況下，數字爲5。然後將任務提交到進程池。

示例

現在將考慮創建線程池時使用的相同示例，唯一的區別是現在將使用ProcessPoolExecutor而不是ThreadPoolExecutor。

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

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

執行上面示例代碼，得到以下結果 -

False
False
Completed

在上面的例子中，一個ProcessPoolExecutor已經被構造成5個線程。然後在提交消息之前等待2秒的任務被提交給進程池執行器。從輸出中可以看出，任務直到2秒才完成，所以第一次調用done()將返回False。2秒後，任務完成，通過調用result()方法得到未來的結果。

實例化ProcessPoolExecutor - 上下文管理器

實例化ProcessPoolExecutor的另一種方法是藉助上下文管理器。它的工作方式與上例中使用的方法類似。使用上下文管理器的主要優點是它在語法上看起來不錯。實例化可以在下面的代碼的幫助下完成 -

with ProcessPoolExecutor(max_workers = 5) as executor

示例

爲了更好地理解，這裏演示創建線程池示例。在這個例子中，我們需要從導入concurrent.futures模塊開始。然後創建一個名爲load_url()的函數，它將加載請求的url。然後使用池中的5個線程創建ProcessPoolExecutor。 ProcessPoolExecutor已被用作上下文管理器。我們可以通過調用result()方法來獲得future的結果。

import concurrent.futures
from concurrent.futures import ProcessPoolExecutor
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()

def main():
   with concurrent.futures.ProcessPoolExecutor(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)))

if __name__ == '__main__':
   main()

上面的Python腳本將生成以下輸出 -

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

使用Executor.map()函數

Python map()函數廣泛用於執行許多任務。一個這樣的任務是對可迭代內的每個元素應用某個函數。同樣，可以將迭代器的所有元素映射到函數，並將這些作爲獨立作業提交給ProcessPoolExecutor。考慮下面的Python腳本示例來理解這一點。

示例

我們將考慮使用Executor.map()函數創建線程池時使用的相同示例。在下面的示例中，map函數用於將square()函數應用於values數組中的每個值。

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

上面的Python腳本將生成以下輸出 -

何時使用ProcessPoolExecutor和ThreadPoolExecutor？

現在我們已經學習了兩個Executor類 - ThreadPoolExecutor和ProcessPoolExecutor，我們需要知道何時使用哪個執行器。需要在受CPU限制的工作負載情況下選擇ProcessPoolExecutor，而在受I/O限制的工作負載情況下則需要選擇ThreadPoolExecutor。

如果使用ProcessPoolExecutor，那麼不需要擔心GIL，因爲它使用多處理。而且，與ThreadPoolExecution相比，執行時間會更少。考慮下面的Python腳本示例來理解這一點。

示例

import time
import concurrent.futures

value = [8000000, 7000000]

def counting(n):
   start = time.time()
   while n > 0:
      n -= 1
   return time.time() - start

def main():
   start = time.time()
   with concurrent.futures.ProcessPoolExecutor() as executor:
      for number, time_taken in zip(value, executor.map(counting, value)):
         print('Start: {} Time taken: {}'.format(number, time_taken))
   print('Total time taken: {}'.format(time.time() - start))

if __name__ == '__main__':
main()

執行上面示例代碼，得到以下結果 -

Start: 8000000 Time taken: 1.5509998798370361
Start: 7000000 Time taken: 1.3259999752044678
Total time taken: 2.0840001106262207

示例 - 使用ThreadPoolExecutor的Python腳本:

import time
import concurrent.futures

value = [8000000, 7000000]

def counting(n):
   start = time.time()
   while n > 0:
      n -= 1
   return time.time() - start

def main():
   start = time.time()
   with concurrent.futures.ThreadPoolExecutor() as executor:
      for number, time_taken in zip(value, executor.map(counting, value)):
         print('Start: {} Time taken: {}'.format(number, time_taken))
      print('Total time taken: {}'.format(time.time() - start))

if __name__ == '__main__':
main()

執行上面示例代碼，得到以下結果 -

Start: 8000000 Time taken: 3.8420000076293945
Start: 7000000 Time taken: 3.6010000705718994
Total time taken: 3.8480000495910645

從上述兩個程序的輸出中，可以看到使用ProcessPoolExecutor和ThreadPoolExecutor時執行時間的差異。

Python併發編程教程

Python模塊 - Concurrent.futures

執行者類