Python中的併發 - 程序池

上一頁
下一頁

程序池的建立和使用方式與執行緒池相同。程序池可以定義為一組預例項化且處於空閒狀態的程序，隨時準備接收工作。當我們需要執行大量任務時，建立程序池比為每個任務例項化新程序更可取。

Python模組 – concurrent.futures

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

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

Executor類

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

  • 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

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

例項化ProcessPoolExecutor – 上下文管理器

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

with ProcessPoolExecutor(max_workers = 5) as executor

示例

為了更好地理解，我們採用了與建立執行緒池時相同的示例。在此示例中，我們需要首先匯入concurrent.futures模組。然後建立一個名為load_url()的函式，該函式將載入請求的URL。然後使用池中的5個執行緒數建立ProcessPoolExecutor。ProcessPoolExecutor已用作上下文管理器。我們可以透過對它呼叫result()方法來獲取期貨的結果。

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指令碼將生成以下輸出

4
9
16
25

何時使用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

Example- Python script with ThreadPoolExecutor:
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時的執行時間差異。

列印頁面
上一頁
下一頁
廣告