concurrent.futures 是 Python 标准库中用于并行编程的高级模块，它提供了一种高级别的接口来管理线程和进程。通过这个模块，你可以轻松地利用多线程和多进程来并行执行任务，进而提高程序的执行效率。
1. concurrent.futures 概述
concurrent.futures 提供了两种执行器类型：
ThreadPoolExecutor：用于管理线程池。
ProcessPoolExecutor：用于管理进程池。
这两种执行器都实现了同样的接口，因此你可以使用相同的代码逻辑来管理线程和进程。
2. 核心 API
2.1 concurrent.futures.Executor
Executor 是一个抽象基类，它定义了任务提交和管理的核心接口。以下是 Executor 提供的主要方法：
submit(fn, *args, **kwargs)：
- 提交一个函数给执行器，函数会在独立的线程或进程中执行。返回一个 Future 对象。
- 场景：当你需要执行一个后台任务并获取结果时使用。
|
1 2 3 4 5 6 7 8 |
from concurrent.futures import ThreadPoolExecutor
def square(n): return n * n
with ThreadPoolExecutor() as executor: future = executor.submit(square, 10) print(future.result()) # 输出: 100 |
提交一个函数给执行器，函数会在独立的线程或进程中执行。返回一个 Future 对象。
场景：当你需要执行一个后台任务并获取结果时使用。
map(func, *iterables, timeout=None, chunksize=1)：
- 将一个函数应用于一个或多个迭代器中的每个元素，并行地执行。类似于内置的 map() 函数，但它会并行执行。
- 场景：当你有一组数据需要并行处理时使用。
|
1 2 3 |
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)：
- 释放执行器资源。如果 wait=True，则会等待所有提交的任务完成；如果 cancel_futures=True，则会取消所有未开始的任务。
- 场景：当你需要优雅地关闭执行器时使用。
|
1 |
executor.shutdown(wait=True) |
2.2 concurrent.futures.Future
Future 对象用于表示异步执行的任务结果。以下是 Future 提供的主要方法：
result(timeout=None)：
- 获取任务的结果，如果任务还未完成，则会等待。你可以设置一个超时时间。
- 场景：当你需要获取异步任务的执行结果时使用。
|
1 |
result = future.result(timeout=5) # 等待最多5秒 |
exception(timeout=None)：
- 如果任务抛出了异常，则返回该异常对象，否则返回 None。
- 场景：当你想处理任务中的异常时使用。
|
1 2 3 4 |
try: result = future.result() except Exception as e: print(f"Error occurred: {e}") |
done()：
- 检查任务是否已完成。
- 场景：当你想知道任务是否已经完成时使用。
|
1 2 |
if future.done(): print("Task is completed.") |
add_done_callback(fn)：
- 为 Future 对象添加一个回调函数，当任务完成时会自动调用此回调。
- 场景：当你需要在任务完成后自动触发某些操作时使用。
|
1 2 3 4 |
def on_done(fut): print(f"Task done with result: {fut.result()}")
future.add_done_callback(on_done) |
3. ThreadPoolExecutor 与 ProcessPoolExecutor
3.1 ThreadPoolExecutor
线程池执行器，用于管理线程。适用于 I/O 密集型任务，如文件操作、网络请求等。
|
1 2 |
with ThreadPoolExecutor(max_workers=5) as executor: future = executor.submit(square, 10) |
参数说明：
max_workers：最大并发线程数。
3.2 ProcessPoolExecutor
进程池执行器，用于管理进程。适用于 CPU 密集型任务，如计算密集型操作。
|
1 2 3 4 |
from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor(max_workers=5) as executor: future = executor.submit(square, 10) |
参数说明：
max_workers：最大并发进程数。
4. 使用场景
4.1 I/O 密集型任务
场景：当你有多个需要等待 I/O 操作（如文件读取、网络请求）的任务时，可以使用 ThreadPoolExecutor 来并行执行这些任务，从而减少总的等待时间。
|
1 2 3 4 5 6 7 8 9 10 11 |
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 密集型任务
场景：当你有多个需要大量计算的任务时，可以使用 ProcessPoolExecutor 来并行执行，从而充分利用多核 CPU 提高效率。
|
1 2 3 4 5 6 7 8 9 |
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. 总结
concurrent.futures 提供了一个方便的接口来管理多线程和多进程的并发执行。通过理解和使用这些 API，你可以更有效地编写并行程序，提高程序的执行效率。在选择使用 ThreadPoolExecutor 还是 ProcessPoolExecutor 时，应根据任务的性质（I/O 密集型或 CPU 密集型）来决定。
