進程池可以像建立和使用執行緒池一樣建立和使用。 進程池可以被定義為一組預先範例化和空閒的進程,它們隨時可以開始工作。 當我們需要執行大量任務時,建立進程池優於為每個任務範例化新進程。
Python標準庫有一個叫做concurrent.futures
的模組。 這個模組是在Python 3.2中新增的,為開發人員提供了啟動非同步任務的高階介面。 它是Python的執行緒和多處理模組的頂層的一個抽象層,用於提供使用執行緒或進程池執行任務的介面。
在後面的章節中,我們將要學習concurrent.futures
模組的不同子類。
Executor是 Python concurrent.futures
模組的抽象類。 它不能直接使用,我們需要使用以下具體子類之一 -
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指令碼將生成以下輸出 -
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
範例 - 使用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
時執行時間的差異。