Python 库手册：concurrent.futures 并发编程模块

MediaTea

1038人浏览 · 2025-07-25 10:00:00

MediaTea · 2025-07-25 10:00:00 发布

concurrent.futures 是 Python 从 3.2 开始引入的一个并发编程标准库，提供高级接口，用于执行异步任务。它基于线程（ThreadPoolExecutor）或进程（ProcessPoolExecutor）池，通过统一的 API 管理多个任务的提交与执行，显著简化了并发编程的复杂。

与传统的 threading 和 multiprocessing 相比，concurrent.futures 更适合管理大量独立任务，如 I/O 密集型请求或 CPU 密集型计算。

常见应用场景：

（1）并发下载多个网页或文件（I/O 密集）。

（2）并行处理大量图像、数据块（CPU 密集）。

（3）批量提交任务并等待结果返回。

（4）实现任务超时、取消、优先级等机制。

（5）用于 GUI 程序或服务器中进行后台任务处理。

◆ ◆ ◆

核心概念

1、Executor（执行器）

concurrent.futures.Executor 是一个抽象基类，定义了任务提交和生命周期管理的接口。

不能直接实例化，通常使用它的两个子类：

ThreadPoolExecutor：基于线程，适用于 I/O 密集型任务。

ProcessPoolExecutor：基于进程，适用于 CPU 密集型任务。

2、Future（未来对象）

表示一个异步执行的任务。

Future 对象可用于：查询任务是否完成、获取任务结果（阻塞或超时）、取消任务以及注册回调函数，等等。

◆ ◆ ◆

应用举例

例 1：使用线程池并发执行 I/O 密集型任务

from concurrent.futures import ThreadPoolExecutorimport time
def download(url):    print(f"开始下载 {url}")    time.sleep(2)  # 模拟 I/O 延迟    return f"{url} 下载完成"
urls = ["url1", "url2", "url3"]
with ThreadPoolExecutor(max_workers=3) as executor:    results = executor.map(download, urls)
for result in results:    print(result)

例 2：使用进程池进行并行计算

from concurrent.futures import ProcessPoolExecutor
# 迭代版本的 Fibonacci，避免栈溢出def fib(n):    a, b = 0, 1    for _ in range(n):        a, b = b, a + b    return a
if __name__ == "__main__":  # 多进程必须加此保护    with ProcessPoolExecutor() as executor:        results = executor.map(fib, [20, 25, 30])    for r in results:        print(r)

例 3：使用 submit + Future 获取任务状态

from concurrent.futures import ThreadPoolExecutorimport time
def task(n):    time.sleep(n)    return f"任务 {n} 秒完成"
executor = ThreadPoolExecutor()future = executor.submit(task, 3)
print("等待任务完成…")print(future.result())  # 会阻塞直到任务完成

例 4：使用 as_completed 获取最快完成的任务

from concurrent.futures import ThreadPoolExecutor, as_completedimport time
def work(x):    time.sleep(x)    return f"{x}秒任务完成"
with ThreadPoolExecutor(max_workers=3) as executor:    futures = [executor.submit(work, i) for i in [3, 1, 2]]    for f in as_completed(futures):        print(f.result())

例 5：设置任务超时与取消

from concurrent.futures import ThreadPoolExecutor, TimeoutErrorimport time
def long_task():    time.sleep(5)    return "完成"
executor = ThreadPoolExecutor()future = executor.submit(long_task)
try:    result = future.result(timeout=2)except TimeoutError:    print("任务超时，准备取消…")    future.cancel()

◆ ◆ ◆

常用函数速览

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

将函数应用于每个迭代项，返回结果迭代器（阻塞等待全部完成）。

参数：

func：要并发执行的函数

*iterables：输入参数迭代器

timeout：超时设置

chunksize：用于 ProcessPoolExecutor 时的数据块大小

返回：生成结果的迭代器

Executor.shutdown(wait=True, cancel_futures=False)

关闭线程/进程池，释放资源。