進(jìn)程池可以像創(chuàng)建和使用線程池一樣創(chuàng)建和使用。 進(jìn)程池可以被定義為一組預(yù)先實(shí)例化和空閑的進(jìn)程，它們隨時(shí)可以開始工作。 當(dāng)我們需要執(zhí)行大量任務(wù)時(shí)，創(chuàng)建進(jìn)程池優(yōu)于為每個(gè)任務(wù)實(shí)例化新進(jìn)程。
Python標(biāo)準(zhǔn)庫(kù)有一個(gè)叫做concurrent.futures的模塊。 這個(gè)模塊是在Python 3.2中添加的，為開發(fā)人員提供了啟動(dòng)異步任務(wù)的高級(jí)接口。 它是Python的線程和多處理模塊的頂層的一個(gè)抽象層，用于提供使用線程或進(jìn)程池運(yùn)行任務(wù)的接口。
在后面的章節(jié)中，我們將要學(xué)習(xí)concurrent.futures模塊的不同子類。
Executor是 Python concurrent.futures模塊的抽象類。 它不能直接使用，我們需要使用以下具體子類之一 -
ProcessPoolExecutor - 一個(gè)具體的子類
它是Executor類的具體子類之一。 它使用多重處理，并且我們獲得提交任務(wù)的過程池。 此池將任務(wù)分配給可用的進(jìn)程并安排它們運(yùn)行。
如何創(chuàng)建一個(gè)ProcessPoolExecutor？
通過concurrent.futures模塊及其具體子類Executor的幫助，可以輕松創(chuàng)建一個(gè)過程池。 為此，需要構(gòu)建一個(gè)ProcessPoolExecutor，其中包含需要的池中的進(jìn)程數(shù)。 默認(rèn)情況下，數(shù)字為5。然后將任務(wù)提交到進(jìn)程池。
示例
現(xiàn)在將考慮創(chuàng)建線程池時(shí)使用的相同示例，唯一的區(qū)別是現(xiàn)在將使用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()
執(zhí)行上面示例代碼，得到以下結(jié)果 -
False
False
Completed
在上面的例子中，一個(gè)ProcessPoolExecutor已經(jīng)被構(gòu)造成5個(gè)線程。 然后在提交消息之前等待2秒的任務(wù)被提交給進(jìn)程池執(zhí)行器。 從輸出中可以看出，任務(wù)直到2秒才完成，所以第一次調(diào)用done()將返回False。2秒后，任務(wù)完成，通過調(diào)用result()方法得到未來的結(jié)果。
實(shí)例化ProcessPoolExecutor - 上下文管理器
實(shí)例化ProcessPoolExecutor的另一種方法是借助上下文管理器。 它的工作方式與上例中使用的方法類似。 使用上下文管理器的主要優(yōu)點(diǎn)是它在語法上看起來不錯(cuò)。 實(shí)例化可以在下面的代碼的幫助下完成 -
with ProcessPoolExecutor(max_workers = 5) as executor
示例
為了更好地理解，這里演示創(chuàng)建線程池示例。 在這個(gè)例子中，我們需要從導(dǎo)入concurrent.futures模塊開始。 然后創(chuàng)建一個(gè)名為load_url()的函數(shù)，它將加載請(qǐng)求的url。 然后使用池中的5個(gè)線程創(chuàng)建ProcessPoolExecutor。 ProcessPoolExecutor已被用作上下文管理器。 我們可以通過調(diào)用result()方法來獲得future的結(jié)果。
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()函數(shù)
Python map()函數(shù)廣泛用于執(zhí)行許多任務(wù)。 一個(gè)這樣的任務(wù)是對(duì)可迭代內(nèi)的每個(gè)元素應(yīng)用某個(gè)函數(shù)。 同樣，可以將迭代器的所有元素映射到函數(shù)，并將這些作為獨(dú)立作業(yè)提交給ProcessPoolExecutor。 考慮下面的Python腳本示例來理解這一點(diǎn)。
示例
我們將考慮使用Executor.map()函數(shù)創(chuàng)建線程池時(shí)使用的相同示例。 在下面的示例中，map函數(shù)用于將square()函數(shù)應(yīng)用于values數(shù)組中的每個(gè)值。
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
何時(shí)使用ProcessPoolExecutor和ThreadPoolExecutor？
現(xiàn)在我們已經(jīng)學(xué)習(xí)了兩個(gè)Executor類 - ThreadPoolExecutor和ProcessPoolExecutor，我們需要知道何時(shí)使用哪個(gè)執(zhí)行器。需要在受CPU限制的工作負(fù)載情況下選擇ProcessPoolExecutor，而在受I/O限制的工作負(fù)載情況下則需要選擇ThreadPoolExecutor。
如果使用ProcessPoolExecutor，那么不需要擔(dān)心GIL，因?yàn)樗褂枚嗵幚怼?而且，與ThreadPoolExecution相比，執(zhí)行時(shí)間會(huì)更少。 考慮下面的Python腳本示例來理解這一點(diǎn)。
示例
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()
執(zhí)行上面示例代碼，得到以下結(jié)果 -
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()
執(zhí)行上面示例代碼，得到以下結(jié)果 -
Start: 8000000 Time taken: 3.8420000076293945
Start: 7000000 Time taken: 3.6010000705718994
Total time taken: 3.8480000495910645
從上述兩個(gè)程序的輸出中，可以看到使用ProcessPoolExecutor和ThreadPoolExecutor時(shí)執(zhí)行時(shí)間的差異。
掃碼關(guān)注鍍金池微信號(hào)
手機(jī)訂課，優(yōu)惠多多