Python進行異步執行的庫有threading（多執行緒）和multiprocessing（多行程），這兩個庫為程式提供了豐富的異步操作，但是如果只是進行一些簡單的異步執行，并不需要用到多復雜的場景，可以考慮使用concurrent.confutures，它提供一些簡單常用的異步執行操作，比如submit和map方法，并且在異步執行完后還可以獲取執行物件的回傳結果，
一、抽象基類：Executor
這是一個抽象基類，后面會介紹的用于執行緒異步執行的ThreadPoolExecutor類和用于行程異步執行的ProcessPoolExecutor類都是Executor的子類，Executor抽象類定義了一些常用的異步操作，在進行多執行緒操作和多行程操作的時候都可以使用這些方法，
1. submit方法
submit(fn, *args, **kwargs)：呼叫可執行物件fn并以fn(*args, **kwargs)的方式執行，回傳一個Future物件，
示例：
# max_workers用于指定pool中最大作業物件數量（執行緒或行程）
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(pow, 323, 1235)
print(future.result()) # result方法可以獲取fn的執行結果
2. map方法
map(func, *iterables, timeout=None, chunksize=1)：這個函式和Python內置的map函式非常相似，但是有以下兩點區別：
- func的執行是異步的，多個func的呼叫是并發呼叫，
- iterables引數是立即執行，而不是延遲執行，
引數說明：
- 回傳值：是一個代表結果的迭代器，可以遍歷這個迭代器以獲得結果，
- timeout：指定超時時間，單位為秒，如果map的
__next__()已經被呼叫，并且Executor.map()在timeout指定秒數后回傳的結果不可用，則會拋出concurrent.futures.TimeoutError， - func：如果func在呼叫程序中拋出了例外，那么在回傳的迭代器中獲取對應結果時這個例外就會被拋出，
- chunksize：可以用在ProcessPoolExecutor，在ThreadPoolExecutor中這個引數則是無效的，chunksize用來指定pool中任務塊的數量，默認只有一個任務塊被用來執行（一個任務塊也可以并發執行多個func），對于較大的iterables，多指定幾個任務塊，可以顯著提升性能（這個引數在Python3.5才加入），
示例：
from concurrent.futures import ThreadPoolExecutor
def func(a):
return 6 / a
with ThreadPoolExecutor() as executor:
result = executor.map(func, [3, 2])
for res in result:
print(res)
'''
列印結果：
2.0
3.0
'''
4. shutdown
shutdown(wait=True, *, cancel_futures=False)：當future物件執行完成后向executor發送一個信號，釋放正在使用的所有資源，但是這個函式之后再呼叫Executor.submit()和Executor.map()的話，則會拋出RuntimeError，
引數說明：
- wait：當wait設定為True時，會一直到所有future物件執行完畢并且executor的所有資源都釋放之后此方法才會回傳，當wait設定為False時，此方法會立即回傳，executor對應的資源則會在所有future物件都執行完畢后才會釋放，但是無論wait的值是什么，executor程式會等到所有future物件都執行完畢后才會退出，
- cancel_futures：當這個引數被設定為True時，所有還未開始執行的future都會被取消，但是無論這個引數的值是什么，正在執行或者已執行完畢的future則不會被取消，（cancel_futures在Python3.9中才加入）
- 注：為避免每次都顯式呼叫shutdown方法，可以使用with語法，它會在結束時自動呼叫shutdown方法，
二、執行緒間的異步：ThreadPoolExecutor
ThreadPoolExecutor是抽象類Executor的子類，使用一個執行緒池來執行所有的異步呼叫，注意，當一個future中呼叫了另一個future時會導致死鎖，
concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())
- max_workers：指定執行緒池中的最大執行緒數，（Python3.5更新：如果沒有指定或者為None，則默認與機器上的處理器個數相同，如果執行緒中側重于I/O而不是CPU計算，則可以在此基礎上乘以5來進行設定，）（Python3.8更新：默認會使用min(32, os.cpu_count() + 4)個執行緒，）
- thread_name_prefix：引數用于設定執行緒名稱的前綴，（Python3.6新增）
- initializer和initargs：指定一個在每個執行緒開始前執行的可呼叫物件initializer，initargs則是傳入此物件的引數，（Python3.7新增）
示例：（來自官方檔案）
import concurrent.futures
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/']
# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
with urllib.request.urlopen(url, timeout=timeout) as conn:
return conn.read()
# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
# Start the load operations and mark each future with its URL
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
# as_completed回傳一個包含future物件的迭代器，但future物件只有執行完畢或被取消才能回傳
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = https://www.cnblogs.com/guyuyun/p/future.result()
except Exception as exc:
print('%r generated an exception: %s' % (url, exc))
else:
print('%r page is %d bytes' % (url, len(data)))
三、行程間的異步：ProcessPoolExecutor
ProcessPoolExecutor是抽象Executor的一個子類，使用一個行程池來執行所有的異步呼叫，它會使用multiprocessing模塊，并且會繞過GIL（全域解釋鎖），但這也意味著只能執行和回傳可picklable的物件，注意：在子行程中再次呼叫另外的Executor物件或Future物件將會導致死鎖，
concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=())
- max_workers：該引數用于指定行程池中的最大行程數，如果沒有指定或者設定為None，則默認會使用與機器的處理器數量相同的數，如果引數小于或等于0，則會拋出ValueError，在Windows系統上，最大執行緒數必須小于等于61，否則拋出ValueError，如果沒有指定，則默認為61（哪怕機器的處理數量大于61），
- mp_context：此引數可以為None或者multiprocessing模塊中的context物件，它將會用于啟動行程，如果沒有指定或者為None，則默認使用multiprocessing模塊中的context物件，（Python3.7更新：mp_context引數允許控制行程池中行程的start_method方法）
- initializer和initargs：指定一個在每個行程開始前執行的可呼叫物件initializer，initargs則是傳入此物件的引數，
- Python3.3更新：如果一個行程被突然終止，則會立即拋出BrokenProcessPool，
示例：（來自官方檔案）
import concurrent.futures
import math
PRIMES = [
112272535095293,
112582705942171,
112272535095293,
115280095190773,
115797848077099,
1099726899285419]
def is_prime(n):
if n < 2:
return False
if n == 2:
return True
if n % 2 == 0:
return False
sqrt_n = int(math.floor(math.sqrt(n)))
for i in range(3, sqrt_n + 1, 2):
if n % i == 0:
return False
return True
def main():
with concurrent.futures.ProcessPoolExecutor() as executor:
for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
print('%d is prime: %s' % (number, prime))
if __name__ == '__main__':
main()
四、Future物件
concurrent.futures.Future：Future物件是對異步執行的封裝，由Executor.submit()創建，
方法說明：
- cancel()：會嘗試取消呼叫，如果呼叫正在執行或者已執行完畢，則該呼叫不會被取消，并會回傳False，否則會取消該呼叫，并回傳True，
- cancelled()：如果呼叫成功取消則回傳True，
- running()：如果呼叫已經開始執行，并且無法被取消，則回傳True，
- done()：如果指呼叫被成功取消，或者已執行完畢，則回傳True，
- result(timeout=None)：回傳呼叫的執行結果，如果呼叫還沒有執行完畢，則會等待指定timeout秒數，如果在timeout時間內沒有執行完畢，則會拋出concurrent.futures.TimeoutError，如果timeout沒有指定或者為None，則會一直等待，如果Future執行完成之前被取消了，則會拋出CancelledError，如果呼叫拋出了例外，則此方法也會拋出同樣的例外，
- exception(timeout=None)：回傳呼叫拋出的例外，如果沒有發生例外，則回傳None，
- add_done_callback(fn)：當future被取消或者執行完畢時fn會被立即呼叫，并且該future會作為fn的唯一引數被傳入，當呼叫中加入多個fn時，則這些fn會以呼叫add_done_callback方法的呼叫順序執行，
以下的一些方法只是用于單元測驗和Executor的實作：
- set_running_or_notify_cancel()：這個方法只能由Executor的實作或者單元測驗在Future物件執行之前呼叫，如果這個方法回傳False，則表示Future物件已被取消，即Future.cancel()已被呼叫并回傳True，即“notify_cancel”生效，如果這個方法回傳True，則Future物件不會被取消并且會將它變成正在運行的狀態，也就是說呼叫Future.running()時將回傳True，即“running”生效，這個方法只能被呼叫一次，并且不能在呼叫Future.set_result()或Future.set_exception()之后進行呼叫，
- set_result(result)：設定Future呼叫的回傳結果，這個方法只能由Executor的實作和單元測驗使用，（Python3.8更新：如果Future已經執行完畢，則會拋出concurrent.futures.InvalidStateError）
- set_exception(exception)：將exception例外設定為Future呼叫的回傳結果，這個方法只能由Executor的實作和單元測驗使用，（Python3.8更新：如果Future已經執行完畢，則會拋出concurrent.futures.InvalidStateError）
五、模塊中的其他函式
concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)
等待由fs指定的所有future物件（可能是由不同的Executor創建）執行完畢，回傳一個由集合構成的二元組，第一個集合為done，包含了所有完成或者已被取消的future物件，第二個集合為not_done，包含了所有的待定或者正在運行的future物件，
- timeout：用來指定最大等待時間（秒），如果沒有指定或者為None，則表示會無限制等待下去，
- return_when：表示這個函式回傳的時機，且只能是以下常量之一：
- FIRST_COMPLETED：當有任何一個future物件執行完畢或者被取消則回傳，
- FIRST_EXCEPTION：當有任何一個future物件執行程序中發生了例外則回傳，如果沒有future發生例外，則相當于ALL_COMPLETED，
- ALL_COMPLETED：當所有future物件都執行完畢或者被取消則回傳，
concurrent.futures.as_completed(fs, timeout=None)
回傳一個由fs指定的future實體的迭代器，當這些future實體執行完畢或者被取消時就會生成對應的future物件，fs指定的future實體如果存在重復的，則這些重復的實體只會回傳一次，并且在呼叫該方法前就執行完成的future實體會被優先回傳，如果在timeout指定的時間內還有future沒有開始執行（呼叫__next__()方法），則會拋出concurrent.futures.TimeoutError，如果timeout沒有指定，則會一直等待下去，
六、Exception類
concurrent.futures.CancelledError：當一個future物件被取消的時候拋出，
concurrent.futures.TimeoutError：當一個future物件的執行時間超出了timeout指定的時間時拋出，
concurrent.futures.BrokenExecutor：派生自RuntimeError，當executor因某些原因被中斷，導致不能submit或者不能執行一個新的任務時會拋出，
concurrent.futures.InvalidStateError：當某個操作在一個當前狀態不允許的future上執行時拋出，
concurrent.futures.thread.BrokenThreadPool：派生自BrokenExecutor，當ThreadPoolExecutor中的一個executor初始化失敗時拋出，
concurrent.futures.process.BrokenProcessPool：派生自BrokenExecutor（以前的名稱為RuntimeError），當ProcessPoolExecutor中一個executor被非正常終止時（如被外部操作殺死時）拋出，
本文大多是從官方檔案直接翻過來的：https://docs.python.org/3/library/concurrent.futures.html
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/266607.html
標籤：Python