Executor 對象?

class concurrent.futures.Executor?

抽象類(lèi)提供異步執行調用方法。要通過(guò)它的子類(lèi)調用，而不是直接調用。

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

類(lèi)似于 map(func, *iterables) 函數，除了以下兩點(diǎn)：

• iterables 是立即執行而不是延遲執行的；

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

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 調用引發(fā)一個(gè)異常，當從迭代器中取回它的值時(shí)這個(gè)異常將被引發(fā)。

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

在 3.5 版更改: 加入 chunksize 參數。

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

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

如果 wait 為 True 則此方法只有在所有待執行的 future 對象完成執行且釋放已分配的資源后才會(huì )返回。 如果 wait 為 False，方法立即返回，所有待執行的 future 對象完成執行后會(huì )釋放已分配的資源。 不管 wait 的值是什么，整個(gè) Python 程序將等到所有待執行的 future 對象完成執行后才退出。

如果 cancel_futures 為 True，此方法將取消所有執行器還未開(kāi)始運行的掛起的 Future。 任何已完成或正在運行的 Future 將不會(huì )被取消，無(wú)論 cancel_futures 的值是什么？

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

如果使用 with 語(yǔ)句，你就可以避免顯式調用這個(gè)方法，它將會(huì )停止 Executor (就好像 Executor.shutdown() 調用時(shí) wait 設為 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 的子類(lèi)，它使用線(xiàn)程池來(lái)異步執行調用。

當回調已關(guān)聯(lián)了一個(gè) Future 然后再等待另一個(gè) Future 的結果時(shí)就會(huì )發(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 子類(lèi)使用最多 max_workers 個(gè)線(xiàn)程的線(xiàn)程池來(lái)異步執行調用。

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

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

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

在 3.7 版更改: 加入 initializer 和*initargs* 參數。

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

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

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 類(lèi)是 Executor 的子類(lèi)，它使用進(jìn)程池來(lái)異步地執行調用。 ProcessPoolExecutor 會(huì )使用 multiprocessing 模塊，這允許它繞過(guò) 全局解釋器鎖 但也意味著(zhù)只可以處理和返回可封存的對象。

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

從可調用對象中調用 Executor 或 Future 的方法提交給 ProcessPoolExecutor 會(huì )導致死鎖。

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

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

initializer 是一個(gè)可選的可調用對象，它會(huì )在每個(gè)工作進(jìn)程啟動(dòng)時(shí)被調用；initargs 是傳給 initializer 的參數元組。 如果 initializer 引發(fā)了異常，則所有當前在等待的任務(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 就會(huì )馬上觸發(fā)。 可預計的行為沒(méi)有定義，但執行器上的操作或它的 future 對象會(huì )被凍結或死鎖。

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

加入 initializer 和*initargs* 參數。

在 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 類(lèi)將可調用對象封裝為異步執行。Future 實(shí)例由 Executor.submit() 創(chuàng )建。

class concurrent.futures.Future?

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

cancel()?

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

cancelled()?

如果調用成功取消返回 True。

running()?

如果調用正在執行而且不能被取消那么返回 True 。

done()?

如果調用已被取消或正常結束那么返回 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ā)。

如果調用引發(fā)了一個(gè)異常，這個(gè)方法也會(huì )引發(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ā)。

如果調用正常完成那么返回 None。

add_done_callback(fn)?

附加可調用 fn 到 future 對象。當 future 對象被取消或完成運行時(shí)，將會(huì )調用 fn，而這個(gè) future 對象將作為它唯一的參數。

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

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

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

set_running_or_notify_cancel()?

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

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

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

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

set_result(result)?

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

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

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

set_exception(exception)?

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

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

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

模塊函數?

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 可以用來(lái)控制返回前最大的等待秒數。 timeout 可以為 int 或 float 類(lèi)型。 如果 timeout 未指定或為 None ，則不限制等待時(shí)間。

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

常量	描述
FIRST_COMPLETED	函數將在任意可等待對象結束或取消時(shí)返回。
FIRST_EXCEPTION	函數將在任意可等待對象因引發(fā)異常而結束時(shí)返回。當沒(méi)有引發(fā)任何異常時(shí)它就相當于 ALL_COMPLETED。
ALL_COMPLETED	函數將在所有可等待對象結束或取消時(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 類(lèi)?

exception concurrent.futures.CancelledError?

future 對象被取消時(shí)會(huì )觸發(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?

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

3.7 新版功能.

exception concurrent.futures.InvalidStateError?

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

3.8 新版功能.

exception concurrent.futures.thread.BrokenThreadPool?

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

3.7 新版功能.

exception concurrent.futures.process.BrokenProcessPool?

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

3.3 新版功能.