concurrent.futures 是 Python 標(biāo)準(zhǔn)庫(kù)中用于并行編程的高級(jí)模塊，它提供了一種高級(jí)別的接口來(lái)管理線(xiàn)程和進(jìn)程。通過(guò)這個(gè)模塊，你可以輕松地利用多線(xiàn)程和多進(jìn)程來(lái)并行執(zhí)行任務(wù)，進(jìn)而提高程序的執(zhí)行效率。

1. concurrent.futures 概述

concurrent.futures 提供了兩種執(zhí)行器類(lèi)型：

ThreadPoolExecutor：用于管理線(xiàn)程池。

ProcessPoolExecutor：用于管理進(jìn)程池。

這兩種執(zhí)行器都實(shí)現(xiàn)了同樣的接口，因此你可以使用相同的代碼邏輯來(lái)管理線(xiàn)程和進(jìn)程。

2. 核心 API

2.1 concurrent.futures.Executor

Executor 是一個(gè)抽象基類(lèi)，它定義了任務(wù)提交和管理的核心接口。以下是 Executor 提供的主要方法：

submit(fn, *args, **kwargs)：

• 提交一個(gè)函數(shù)給執(zhí)行器，函數(shù)會(huì)在獨(dú)立的線(xiàn)程或進(jìn)程中執(zhí)行。返回一個(gè) Future 對(duì)象。
• 場(chǎng)景：當(dāng)你需要執(zhí)行一個(gè)后臺(tái)任務(wù)并獲取結(jié)果時(shí)使用。

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor() as executor:
    future = executor.submit(square, 10)
    print(future.result())  # 輸出: 100

提交一個(gè)函數(shù)給執(zhí)行器，函數(shù)會(huì)在獨(dú)立的線(xiàn)程或進(jìn)程中執(zhí)行。返回一個(gè) Future 對(duì)象。

場(chǎng)景：當(dāng)你需要執(zhí)行一個(gè)后臺(tái)任務(wù)并獲取結(jié)果時(shí)使用。

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

• 將一個(gè)函數(shù)應(yīng)用于一個(gè)或多個(gè)迭代器中的每個(gè)元素，并行地執(zhí)行。類(lèi)似于內(nèi)置的 map() 函數(shù)，但它會(huì)并行執(zhí)行。
• 場(chǎng)景：當(dāng)你有一組數(shù)據(jù)需要并行處理時(shí)使用。

with ThreadPoolExecutor() as executor:
    results = executor.map(square, range(10))
    print(list(results))  # 輸出: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

shutdown(wait=True, cancel_futures=False)：

• 釋放執(zhí)行器資源。如果 wait=True，則會(huì)等待所有提交的任務(wù)完成；如果 cancel_futures=True，則會(huì)取消所有未開(kāi)始的任務(wù)。
• 場(chǎng)景：當(dāng)你需要優(yōu)雅地關(guān)閉執(zhí)行器時(shí)使用。

executor.shutdown(wait=True)

2.2 concurrent.futures.Future

Future 對(duì)象用于表示異步執(zhí)行的任務(wù)結(jié)果。以下是 Future 提供的主要方法：

result(timeout=None)：

• 獲取任務(wù)的結(jié)果，如果任務(wù)還未完成，則會(huì)等待。你可以設(shè)置一個(gè)超時(shí)時(shí)間。
• 場(chǎng)景：當(dāng)你需要獲取異步任務(wù)的執(zhí)行結(jié)果時(shí)使用。

result = future.result(timeout=5)  # 等待最多5秒

exception(timeout=None)：

• 如果任務(wù)拋出了異常，則返回該異常對(duì)象，否則返回 None。
• 場(chǎng)景：當(dāng)你想處理任務(wù)中的異常時(shí)使用。

try:
    result = future.result()
except Exception as e:
    print(f"Error occurred: {e}")

done()：

• 檢查任務(wù)是否已完成。
• 場(chǎng)景：當(dāng)你想知道任務(wù)是否已經(jīng)完成時(shí)使用。

if future.done():
    print("Task is completed.")

add_done_callback(fn)：

• 為 Future 對(duì)象添加一個(gè)回調(diào)函數(shù)，當(dāng)任務(wù)完成時(shí)會(huì)自動(dòng)調(diào)用此回調(diào)。
• 場(chǎng)景：當(dāng)你需要在任務(wù)完成后自動(dòng)觸發(fā)某些操作時(shí)使用。

def on_done(fut):
    print(f"Task done with result: {fut.result()}")

future.add_done_callback(on_done)

3. ThreadPoolExecutor 與 ProcessPoolExecutor

3.1 ThreadPoolExecutor

線(xiàn)程池執(zhí)行器，用于管理線(xiàn)程。適用于 I/O 密集型任務(wù)，如文件操作、網(wǎng)絡(luò)請(qǐng)求等。

with ThreadPoolExecutor(max_workers=5) as executor:
    future = executor.submit(square, 10)

參數(shù)說(shuō)明：

max_workers：最大并發(fā)線(xiàn)程數(shù)。

3.2 ProcessPoolExecutor

進(jìn)程池執(zhí)行器，用于管理進(jìn)程。適用于 CPU 密集型任務(wù)，如計(jì)算密集型操作。

from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor(max_workers=5) as executor:
    future = executor.submit(square, 10)

參數(shù)說(shuō)明：

max_workers：最大并發(fā)進(jìn)程數(shù)。

4. 使用場(chǎng)景

4.1 I/O 密集型任務(wù)

場(chǎng)景：當(dāng)你有多個(gè)需要等待 I/O 操作（如文件讀取、網(wǎng)絡(luò)請(qǐng)求）的任務(wù)時(shí)，可以使用 ThreadPoolExecutor 來(lái)并行執(zhí)行這些任務(wù)，從而減少總的等待時(shí)間。

import requests

def fetch_url(url):
    response = requests.get(url)
    return response.status_code

urls = ['https://www.example.com', 'https://www.google.com', 'https://www.github.com']

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(fetch_url, urls)
    print(list(results))

4.2 CPU 密集型任務(wù)

場(chǎng)景：當(dāng)你有多個(gè)需要大量計(jì)算的任務(wù)時(shí)，可以使用 ProcessPoolExecutor 來(lái)并行執(zhí)行，從而充分利用多核 CPU 提高效率。

def fibonacci(n):
    if n <= 1:
        return n
    else:
        return fibonacci(n-1) + fibonacci(n-2)

with ProcessPoolExecutor(max_workers=3) as executor:
    results = executor.map(fibonacci, range(10, 20))
    print(list(results))

5. 總結(jié)

concurrent.futures 提供了一個(gè)方便的接口來(lái)管理多線(xiàn)程和多進(jìn)程的并發(fā)執(zhí)行。通過(guò)理解和使用這些 API，你可以更有效地編寫(xiě)并行程序，提高程序的執(zhí)行效率。在選擇使用 ThreadPoolExecutor 還是 ProcessPoolExecutor 時(shí)，應(yīng)根據(jù)任務(wù)的性質(zhì)（I/O 密集型或 CPU 密集型）來(lái)決定。

到此這篇關(guān)于python中concurrent.futures的具體使用的文章就介紹到這了,更多相關(guān)python concurrent.futures內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論