新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案

創(chuàng)新互聯(lián)Python教程：concurrent.futures—-啟動并行任務(wù)

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

3.2 新版功能.

成都創(chuàng)新互聯(lián)公司主營阿里地區(qū)網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營網(wǎng)站建設(shè)方案,App定制開發(fā),阿里地區(qū)h5成都微信小程序搭建,阿里地區(qū)網(wǎng)站營銷推廣歡迎阿里地區(qū)等地區(qū)企業(yè)咨詢

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

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

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

Availability: not Emscripten, not WASI.

This module does not work or is not available on WebAssembly platforms wasm32-emscripten and wasm32-wasi. See WebAssembly platforms for more information.

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í)行并返回一個代表該可調(diào)用對象的執(zhí)行的 Future 對象。
    
    
    
    
     
     
     
     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 的多個調(diào)用可以并發(fā)執(zhí)行。
  
  
  
  
   
   
   
   The returned iterator raises a [TimeoutError]($e69e6dad35dc3610.md#TimeoutError "TimeoutError") if [\_\_next\_\_()]($c2b73e6dbd6fbe32.md#iterator.__next__ "iterator.__next__") is called and the result isn't available after *timeout* seconds from the original call to [Executor.map()](#concurrent.futures.Executor.map "concurrent.futures.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ā)一個異常，當(dāng)從迭代器中取回它的值時這個異常將被引發(fā)。   
   
   
      
   
   
   使用 [ProcessPoolExecutor](#concurrent.futures.ProcessPoolExecutor "concurrent.futures.ProcessPoolExecutor") 時，這個方法會將 *iterables* 分割任務(wù)塊并作為獨(dú)立的任務(wù)并提交到執(zhí)行池中。這些塊的大概數(shù)量可以由 *chunksize* 指定正整數(shù)設(shè)置。 對很長的迭代器來說，使用大的 *chunksize* 值比默認(rèn)值 1 能顯著地提高性能。 *chunksize* 對 [ThreadPoolExecutor](#concurrent.futures.ThreadPoolExecutor "concurrent.futures.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 的值是什么，整個 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)用這個方法，它將會停止 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')    
    
    
    
在 3.9 版更改: 增加了 cancel_futures。

ThreadPoolExecutor

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

當(dāng)可調(diào)用對象已關(guān)聯(lián)了一個 Future 然后在等待另一個 Future 的結(jié)果時就會導(dǎo)致死鎖情況。例如:

 
 
 
 
  
  
  
  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)用。

All threads enqueued to ThreadPoolExecutor will be joined before the interpreter can exit. Note that the exit handler which does this is executed before any exit handlers added using atexit. This means exceptions in the main thread must be caught and handled in order to signal threads to exit gracefully. For this reason, it is recommended that ThreadPoolExecutor not be used for long-running tasks.

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

在 3.5 版更改: 如果 max_workers 為 None 或沒有指定，將默認(rèn)為機(jī)器處理器的個數(shù)，假如 ThreadPoolExecutor 則重于I/O操作而不是CPU運(yùn)算，那么可以乘以 5 ，同時工作線程的數(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)。這個默認(rèn)值會保留至少 5 個工作線程用于 I/O 密集型任務(wù)。對于那些釋放了 GIL 的 CPU 密集型任務(wù)，它最多會使用 32 個 CPU 核心。這樣能夠避免在多核機(jī)器上不知不覺地使用大量資源。

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

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)程池來異步地執(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 個進(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。它將被用來啟動工作進(jìn)程。如果 mp_context 為 None 或未給出，則將使用默認(rèn)的多進(jìn)程上下文。

initializer 是一個可選的可調(diào)用對象，它會在每個工作進(jìn)程啟動時被調(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 absence of a mp_context parameter. This feature is incompatible with the “fork” start method.

在 3.3 版更改: 如果其中一個工作進(jìn)程被突然終止，BrokenProcessPool 就會馬上觸發(fā)。可預(yù)計的行為沒有定義，但執(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.

ProcessPoolExecutor 例子

 
 
 
 
  
  
  
  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ā)了一個異常，這個方法也會引發(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)行時，將會調(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 實(shí)現(xiàn)。

set_running_or_notify_cancel()
這個方法只可以在執(zhí)行關(guān)聯(lián) Future 工作之前由 Executor 實(shí)現(xiàn)調(diào)用或由單測試調(diào)用。
If the method returns False then the Future was cancelled, i.e. Future.cancel() was called and returned True. Any threads waiting on the Future completing (i.e. through as_completed() or wait()) will be woken up.
If the method returns True then the Future was not cancelled and has been put in the running state, i.e. calls to Future.running() will return 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 實(shí)現(xiàn)和單元測試使用。
在 3.8 版更改: 如果 Future 已經(jīng)完成則此方法會引發(fā) concurrent.futures.InvalidStateError。

set_exception(exception)
設(shè)置 Future 關(guān)聯(lián)工作的結(jié)果給 Exception exception 。
這個方法只可以由 Executor 實(shí)現(xiàn)和單元測試使用。
在 3.8 版更改: 如果 Future 已經(jīng)完成則此方法會引發(fā) concurrent.futures.InvalidStateError。

模塊函數(shù)

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

等待由 fs 指定的 Future 實(shí)例（可能由不同的 Executor 實(shí)例創(chuàng)建）完成。重復(fù)傳給 fs 的 future 會被移除并將只返回一次。返回一個由集合組成的具名 2 元組。第一個集合的名稱為 done，包含在等待完成之前已完成的 future（包括正常結(jié)束或被取消的 future）。第二個集合的名稱為 not_done，包含未完成的 future（包括掛起的或正在運(yùn)行的 future）。

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

return_when 指定此函數(shù)應(yīng)在何時返回。它必須為以下常數(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)

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.

參見

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

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

Exception 類

exception concurrent.futures.CancelledError

future 對象被取消時會觸發(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ù)時就會被引發(fā)派生于 RuntimeError 的異常類。

3.7 新版功能.

exception concurrent.futures.InvalidStateError

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

3.8 新版功能.

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 新版功能.

分享標(biāo)題：創(chuàng)新互聯(lián)Python教程：concurrent.futures—-啟動并行任務(wù)
文章轉(zhuǎn)載：http://www.fisionsoft.com.cn/article/cdjsojc.html