concurrent.futures --- 啟動(dòng)并行任務(wù)?

Executor 對象?

class concurrent.futures.Executor?

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

submit(fn, /, *args, **kwargs)?

Schedules the callable, fn, to be executed as fn(*args, **kwargs) and returns a Future object representing the execution of the callable.

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ù)，除了以下兩點(diǎn)：

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

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

The returned iterator raises a TimeoutError if __next__() is called and the result isn't available after timeout seconds from the original call to Executor.map(). timeout can be an int or a float. If timeout is not specified or None, there is no limit to the wait time.

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

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

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

shutdown(wait=True, *, cancel_futures=False)?

當(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 的值是什么，整個(gè) Python 程序?qū)⒌鹊剿写龍?zhí)行的 future 對象完成執(zhí)行后才退出。

如果 cancel_futures 為 True，此方法將取消所有執(zhí)行器還未開始運(yùn)行的掛起的 Future。 任何已完成或正在運(yùn)行的 Future 將不會被取消，無論 cancel_futures 的值是什么？

如果 cancel_futures 和 wait 均為 True，則執(zhí)行器已開始運(yùn)行的所有 Future 將在此方法返回之前完成。 其余的 Future 會被取消。

如果使用 with 語句，你就可以避免顯式調(diào)用這個(gè)方法，它將會停止 Executor (就好像 Executor.shutdown() 調(diào)用時(shí) 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')

在 3.9 版更改: 增加了 cancel_futures。

ThreadPoolExecutor?

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

當(dāng)回調(diào)已關(guān)聯(lián)了一個(gè) Future 然后再等待另一個(gè) Future 的結(jié)果時(shí)就會發(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 個(gè)線程的線程池來異步執(zhí)行調(diào)用。

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

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

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

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

在 3.8 版更改: max_workers 的默認(rèn)值已改為 min(32, os.cpu_count() + 4)。 這個(gè)默認(rèn)值會保留至少 5 個(gè)工作線程用于 I/O 密集型任務(wù)。 對于那些釋放了 GIL 的 CPU 密集型任務(wù)，它最多會使用 32 個(gè) CPU 核心。這樣能夠避免在多核機(jī)器上不知不覺地使用大量資源。

現(xiàn)在 ThreadPoolExecutor 在啟動(dòng) max_workers 個(gè)工作線程之前也會重用空閑的工作線程。

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)程池來異步地執(zhí)行調(diào)用。 ProcessPoolExecutor 會使用 multiprocessing 模塊，這允許它繞過 全局解釋器鎖 但也意味著只可以處理和返回可封存的對象。

__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=(), max_tasks_per_child=None)?

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

initializer 是一個(gè)可選的可調(diào)用對象，它會在每個(gè)工作進(jìn)程啟動(dòng)時(shí)被調(diào)用；initargs 是傳給 initializer 的參數(shù)元組。 如果 initializer 引發(fā)了異常，則所有當(dāng)前在等待的任務(wù)以及任何向進(jìn)程池提交更多任務(wù)的嘗試都將引發(fā) BrokenProcessPool。

max_tasks_per_child is an optional argument that specifies the maximum number of tasks a single process can execute before it will exit and be replaced with a fresh worker process. By default max_tasks_per_child is None which means worker processes will live as long as the pool. When a max is specified, the "spawn" multiprocessing start method will be used by default in absense of a mp_context parameter. This feature is incompatible with the "fork" start method.

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

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

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

在 3.11 版更改: The max_tasks_per_child argument was added to allow users to control the lifetime of workers in the pool.

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 對象?

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

class concurrent.futures.Future?

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

cancel()?

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

cancelled()?

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

running()?

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

done()?

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

result(timeout=None)?

Return the value returned by the call. If the call hasn't yet completed then this method will wait up to timeout seconds. If the call hasn't completed in timeout seconds, then a TimeoutError will be raised. timeout can be an int or float. If timeout is not specified or None, there is no limit to the wait time.

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

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

exception(timeout=None)?

Return the exception raised by the call. If the call hasn't yet completed then this method will wait up to timeout seconds. If the call hasn't completed in timeout seconds, then a TimeoutError will be raised. timeout can be an int or float. If timeout is not specified or None, there is no limit to the wait time.

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

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

add_done_callback(fn)?

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

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

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

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

set_running_or_notify_cancel()?

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

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

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

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

set_result(result)?

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

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

在 3.8 版更改: 如果 Future 已經(jīng)完成則此方法會引發(fā) concurrent.futures.InvalidStateError。

set_exception(exception)?

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

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

在 3.8 版更改: 如果 Future 已經(jīng)完成則此方法會引發(fā) concurrent.futures.InvalidStateError。

模塊函數(shù)?

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

Wait for the Future instances (possibly created by different Executor instances) given by fs to complete. Duplicate futures given to fs are removed and will be returned only once. Returns a named 2-tuple of sets. The first set, named done, contains the futures that completed (finished or cancelled futures) before the wait completed. The second set, named not_done, contains the futures that did not complete (pending or running futures).

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

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

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

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

Returns an iterator over the Future instances (possibly created by different Executor instances) given by fs that yields futures as they complete (finished or cancelled futures). Any futures given by fs that are duplicated will be returned once. Any futures that completed before as_completed() is called will be yielded first. The returned iterator raises a TimeoutError if __next__() is called and the result isn't available after timeout seconds from the original call to as_completed(). timeout can be an int or float. If timeout is not specified or None, there is no limit to the wait time.

Exception 類?

exception concurrent.futures.CancelledError?

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

exception concurrent.futures.TimeoutError?

A deprecated alias of TimeoutError, raised when a future operation exceeds the given timeout.

在 3.11 版更改: This class was made an alias of TimeoutError.

exception concurrent.futures.BrokenExecutor?

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

3.7 新版功能.

exception concurrent.futures.InvalidStateError?

當(dāng)某個(gè)操作在一個(gè)當(dāng)前狀態(tài)所不允許的 future 上執(zhí)行時(shí)將被引發(fā)。

3.8 新版功能.

exception concurrent.futures.thread.BrokenThreadPool?

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

3.7 新版功能.

exception concurrent.futures.process.BrokenProcessPool?

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

3.3 新版功能.