concurrent.futures --- 啟動并行任務(wù)

3.2 新版功能.

源碼: Lib/concurrent/futures/thread.py 和 Lib/concurrent/futures/process.py

---

concurrent.futures 模塊提供異步執(zhí)行可調(diào)用對象高層接口。

異步執(zhí)行可以由 ThreadPoolExecutor 使用線程或由 ProcessPoolExecutor 使用單獨的進(jìn)程來實現(xiàn)。 兩者都是實現(xiàn)抽像類 Executor 定義的接口。

Executor 對象

class concurrent.futures.Executor

抽象類提供異步執(zhí)行調(diào)用方法。要通過它的子類調(diào)用，而不是直接調(diào)用。

submit(fn, *args, **kwargs)

調(diào)度可調(diào)用對象 fn，以 fn(*args **kwargs) 方式執(zhí)行并返回 Future 對像代表可調(diào)用對象的執(zhí)行。:

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print(future.result())

map(func, *iterables, timeout=None, chunksize=1)

類似于 map(func, *iterables) 函數(shù)，除了以下兩點：

• iterables 是立即執(zhí)行而不是延遲執(zhí)行的；

• func 是異步執(zhí)行的，對 func 的多個調(diào)用可以并發(fā)執(zhí)行。

如果從原始調(diào)用到 Executor.map() 經(jīng)過 timeout 秒后， __next__() 已被調(diào)用且返回的結(jié)果還不可用，那么已返回的迭代器將觸發(fā) concurrent.futures.TimeoutError 。 timeout 可以是整數(shù)或浮點數(shù)。如果 timeout 沒有指定或為 None ，則沒有超時限制。

如果 func 調(diào)用引發(fā)一個異常，當(dāng)從迭代器中取回它的值時這個異常將被引發(fā)。

使用 ProcessPoolExecutor 時，這個方法會將 iterables 分割任務(wù)塊并作為獨立的任務(wù)并提交到執(zhí)行池中。這些塊的大概數(shù)量可以由 chunksize 指定正整數(shù)設(shè)置。 對很長的迭代器來說，使用大的 chunksize 值比默認(rèn)值 1 能顯著地提高性能。 chunksize 對 ThreadPoolExecutor 沒有效果。

在 3.5 版更改: 加入 chunksize 參數(shù)。

shutdown(wait=True)

當(dāng)待執(zhí)行的 future 對象完成執(zhí)行后向執(zhí)行者發(fā)送信號，它就會釋放正在使用的任何資源。 在關(guān)閉后調(diào)用 Executor.submit() 和 Executor.map() 將會引發(fā) RuntimeError。

如果 wait 為 True 則此方法只有在所有待執(zhí)行的 future 對象完成執(zhí)行且釋放已分配的資源后才會返回。 如果 wait 為 False，方法立即返回，所有待執(zhí)行的 future 對象完成執(zhí)行后會釋放已分配的資源。 不管 wait 的值是什么，整個 Python 程序?qū)⒌鹊剿写龍?zhí)行的 future 對象完成執(zhí)行后才退出。

如果使用 with 語句，你就可以避免顯式調(diào)用這個方法，它將會停止 Executor (就好像 Executor.shutdown() 調(diào)用時 wait 設(shè)為 True 一樣等待):

import shutil
with ThreadPoolExecutor(max_workers=4) as e:
    e.submit(shutil.copy, 'src1.txt', 'dest1.txt')
    e.submit(shutil.copy, 'src2.txt', 'dest2.txt')
    e.submit(shutil.copy, 'src3.txt', 'dest3.txt')
    e.submit(shutil.copy, 'src4.txt', 'dest4.txt')

ThreadPoolExecutor

ThreadPoolExecutor 是 Executor 的子類，它使用線程池來異步執(zhí)行調(diào)用。

當(dāng)回調(diào)已關(guān)聯(lián)了一個 Future 然后再等待另一個 Future 的結(jié)果時就會發(fā)產(chǎn)死鎖情況。例如:

import time
def wait_on_b():
    time.sleep(5)
    print(b.result())  # b will never complete because it is waiting on a.
    return 5

def wait_on_a():
    time.sleep(5)
    print(a.result())  # a will never complete because it is waiting on b.
    return 6


executor = ThreadPoolExecutor(max_workers=2)
a = executor.submit(wait_on_b)
b = executor.submit(wait_on_a)

與:

def wait_on_future():
    f = executor.submit(pow, 5, 2)
    # This will never complete because there is only one worker thread and
    # it is executing this function.
    print(f.result())

executor = ThreadPoolExecutor(max_workers=1)
executor.submit(wait_on_future)

class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())

Executor 子類使用最多 max_workers 個線程的線程池來異步執(zhí)行調(diào)用。

initializer 是在每個工作者線程開始處調(diào)用的一個可選可調(diào)用對象。 initargs 是傳遞給初始化器的元組參數(shù)。任何向池提交更多工作的嘗試， initializer 都將引發(fā)一個異常，當(dāng)前所有等待的工作都會引發(fā)一個 BrokenThreadPool。

在 3.5 版更改: 如果 max_workers 為 None 或沒有指定，將默認(rèn)為機(jī)器處理器的個數(shù)，假如 ThreadPoolExecutor 側(cè)重于 I/O 操作而不是 CPU 運算，那么可以乘以 5，同時工作線程的數(shù)量可以比 ProcessPoolExecutor 的數(shù)量高。

3.6 新版功能: 添加 thread_name_prefix 參數(shù)允許用戶控制由線程池創(chuàng)建的 threading.Thread 工作線程名稱以方便調(diào)試。

在 3.7 版更改: 加入 initializer 和*initargs* 參數(shù)。

ThreadPoolExecutor 例子

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}
    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)))

ProcessPoolExecutor

ProcessPoolExecutor 是 Executor 的子類，它使用進(jìn)程池來實現(xiàn)異步執(zhí)行調(diào)用。 ProcessPoolExecutor 使用 multiprocessing 回避 Global Interpreter Lock 但也意味著只可以處理和返回可序列化的對象。

__main__ 模塊必須可以被工作者子進(jìn)程導(dǎo)入。這意味著 ProcessPoolExecutor 不可以工作在交互式解釋器中。

從可調(diào)用對象中調(diào)用 Executor 或 Future 的方法提交給 ProcessPoolExecutor 會導(dǎo)致死鎖。

class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=())

異步執(zhí)行調(diào)用的 Executor 子類使用一個最多有 max_workers 個進(jìn)程的進(jìn)程池。 如果 max_workers 為 None 或未給出，它將默認(rèn)為機(jī)器的處理器個數(shù)。 如果 max_workers 小于等于 0，則將引發(fā) ValueError。 在 Windows 上，max_workers 必須小于等于 61，否則將引發(fā) ValueError。 如果 max_workers 為 None，則所選擇的默認(rèn)最多為 61，即使存在更多處理器。 mp_context 可以是一個多進(jìn)程上下文或是 None。 它將被用來啟動工作者。 如果 mp_context 為 None 或未給出，將使用默認(rèn)的多進(jìn)程上下文。

initializer 是在每個工作者進(jìn)程開始處調(diào)用的一個可選可調(diào)用對象。 initargs 是傳遞給初始化器的元組參數(shù)。任何向池提交更多工作的嘗試， initializer 都將引發(fā)一個異常，當(dāng)前所有等待的工作都會引發(fā)一個 BrokenProcessPool。

在 3.3 版更改: 如果其中一個工作進(jìn)程被突然終止，BrokenProcessPool 就會馬上觸發(fā)。 可預(yù)計的行為沒有定義，但執(zhí)行器上的操作或它的 future 對象會被凍結(jié)或死鎖。

在 3.7 版更改: 添加 mp_context 參數(shù)允許用戶控制由進(jìn)程池創(chuàng)建給工作者進(jìn)程的開始方法 。

加入 initializer 和*initargs* 參數(shù)。

ProcessPoolExecutor 例子

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    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 對象

Future 類將可調(diào)用對象封裝為異步執(zhí)行。Future 實例由 Executor.submit() 創(chuàng)建。

class concurrent.futures.Future

將可調(diào)用對象封裝為異步執(zhí)行。Future 實例由 Executor.submit() 創(chuàng)建，除非測試，不應(yīng)直接創(chuàng)建。

cancel()

嘗試取消調(diào)用。 如果調(diào)用正在執(zhí)行或已結(jié)束運行不能被取消則該方法將返回 False，否則調(diào)用會被取消并且該方法將返回 True。

cancelled()

如果調(diào)用成功取消返回 True。

running()

如果調(diào)用正在執(zhí)行而且不能被取消那么返回 True 。

done()

如果調(diào)用已被取消或正常結(jié)束那么返回 True。

result(timeout=None)

返回調(diào)用返回的值。如果調(diào)用還沒完成那么這個方法將等待 timeout 秒。如果在 timeout 秒內(nèi)沒有執(zhí)行完成，concurrent.futures.TimeoutError 將會被觸發(fā)。timeout 可以是整數(shù)或浮點數(shù)。如果 timeout 沒有指定或為 None，那么等待時間就沒有限制。

如果 futrue 在完成前被取消則 CancelledError 將被觸發(fā)。

如果調(diào)用引發(fā)了一個異常，這個方法也會引發(fā)同樣的異常。

exception(timeout=None)

返回由調(diào)用引發(fā)的異常。如果調(diào)用還沒完成那么這個方法將等待 timeout 秒。如果在 timeout 秒內(nèi)沒有執(zhí)行完成，concurrent.futures.TimeoutError 將會被觸發(fā)。timeout 可以是整數(shù)或浮點數(shù)。如果 timeout 沒有指定或為 None，那么等待時間就沒有限制。

如果 futrue 在完成前被取消則 CancelledError 將被觸發(fā)。

如果調(diào)用正常完成那么返回 None。

add_done_callback(fn)

附加可調(diào)用 fn 到 future 對象。當(dāng) future 對象被取消或完成運行時，將會調(diào)用 fn，而這個 future 對象將作為它唯一的參數(shù)。

加入的可調(diào)用對象總被屬于添加它們的進(jìn)程中的線程按加入的順序調(diào)用。如果可調(diào)用對象引發(fā)一個 Exception 子類，它會被記錄下來并被忽略掉。如果可調(diào)用對象引發(fā)一個 BaseException 子類，這個行為沒有定義。

如果 future 對象已經(jīng)完成或已取消，fn 會被立即調(diào)用。

下面這些 Future 方法用于單元測試和 Executor 實現(xiàn)。

set_running_or_notify_cancel()

這個方法只可以在執(zhí)行關(guān)聯(lián) Future 工作之前由 Executor 實現(xiàn)調(diào)用或由單測試調(diào)用。

如果這個方法返回 False 那么 Future 已被取消，即 Future.cancel() 已被調(diào)用并返回 True 。等待 Future 完成 (即通過 as_completed() 或 wait()) 的線程將被喚醒。

如果這個方法返回 True 那么 Future 不會被取消并已將它變?yōu)檎谶\行狀態(tài)，也就是說調(diào)用 Future.running() 時將返回 True。

這個方法只可以被調(diào)用一次并且不能在調(diào)用 Future.set_result() 或 Future.set_exception() 之后再調(diào)用。

set_result(result)

設(shè)置將 Future 關(guān)聯(lián)工作的結(jié)果給 result 。

這個方法只可以由 Executor 實現(xiàn)和單元測試使用。

set_exception(exception)

設(shè)置 Future 關(guān)聯(lián)工作的結(jié)果給 Exception exception 。

這個方法只可以由 Executor 實現(xiàn)和單元測試使用。

模塊函數(shù)

concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)

等待 fs 指定的 Future 實例（可能由不同的 Executor 實例創(chuàng)建）完成。 返回一個由集合構(gòu)成的具名 2 元組。 第一個集合名稱為 done，包含在等待完成之前已完成的期程（包括正常結(jié)束或被取消的 future 對象）。 第二個集合名稱為 not_done，包含未完成的 future 對象（包括掛起的或正在運行的 future 對象）。

timeout 可以用來控制返回前最大的等待秒數(shù)。 timeout 可以為 int 或 float 類型。 如果 timeout 未指定或為 None ，則不限制等待時間。

return_when 指定此函數(shù)應(yīng)在何時返回。它必須為以下常數(shù)之一:

常數(shù)	描述
FIRST_COMPLETED	函數(shù)將在任意可等待對象結(jié)束或取消時返回。
FIRST_EXCEPTION	函數(shù)將在任意可等待對象因引發(fā)異常而結(jié)束時返回。當(dāng)沒有引發(fā)任何異常時它就相當(dāng)于 ALL_COMPLETED。
ALL_COMPLETED	函數(shù)將在所有可等待對象結(jié)束或取消時返回。

concurrent.futures.as_completed(fs, timeout=None)

返回一個包含 fs 所指定的 Future 實例（可能由不同的 Executor 實例創(chuàng)建）的迭代器，這些實例會在完成時生成 future 對象（包括正常結(jié)束或被取消的 future 對象）。 任何由 fs 所指定的重復(fù) future 對象將只被返回一次。 任何在 as_completed() 被調(diào)用之前完成的 future 對象將優(yōu)先被生成。 如果 __next__() 被調(diào)用并且在對 as_completed() 的原始調(diào)用 timeout 秒之后結(jié)果仍不可用，則返回的迭代器將引發(fā) concurrent.futures.TimeoutError。 timeout 可以為整數(shù)或浮點數(shù)。 如果 timeout 未指定或為 None，則不限制等待時間。

參見

PEP 3148 -- future 對象 - 異步執(zhí)行指令。

該提案描述了Python標(biāo)準(zhǔn)庫中包含的這個特性。

Exception類

exception concurrent.futures.CancelledError

future 對象被取消時會觸發(fā)。

exception concurrent.futures.TimeoutError

future 對象執(zhí)行超出給定的超時數(shù)值時引發(fā)。

exception concurrent.futures.BrokenExecutor

當(dāng)執(zhí)行器被某些原因中斷而且不能用來提交或執(zhí)行新任務(wù)時就會被引發(fā)派生于 RuntimeError 的異常類。

3.7 新版功能.

exception concurrent.futures.thread.BrokenThreadPool

當(dāng) ThreadPoolExecutor 中的其中一個工作者初始化失敗時會引發(fā)派生于 BrokenExecutor 的異常類。

3.7 新版功能.

exception concurrent.futures.process.BrokenProcessPool

當(dāng) ThreadPoolExecutor 中的其中一個工作者不完整終止時(比如，被外部殺死)會引發(fā)派生于 BrokenExecutor ( 原名 RuntimeError ) 的異常類。

3.3 新版功能.