concurrent.futures：统一并发接口

concurrent.futures 提供高层 API，统一线程池和进程池的使用方式。无需直接管理线程或进程，专注于任务提交和结果获取。

Executor 接口

两种执行器共享相同接口：

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def task(n):
    return n * n

# 线程池
with ThreadPoolExecutor(max_workers=4) as executor:
    result = executor.submit(task, 5)
    print(result.result())  # 25

# 进程池
with ProcessPoolExecutor(max_workers=4) as executor:
    result = executor.submit(task, 5)
    print(result.result())  # 25

ThreadPoolExecutor

适合 I/O 密集型任务：

from concurrent.futures import ThreadPoolExecutor
import time

def download(url):
    print(f"下载 {url}")
    time.sleep(2)  # 模拟网络请求
    return f"{url} 完成"

urls = ["url1", "url2", "url3", "url4"]

# 自动管理线程
with ThreadPoolExecutor(max_workers=2) as executor:
    results = executor.map(download, urls)
    
    for result in results:
        print(result)

输出示例（耗时约4秒，两批并行）：

下载 url1
下载 url2
（2秒后）
url1 完成
url2 完成
下载 url3
下载 url4
（2秒后）
url3 完成
url4 完成

ProcessPoolExecutor

适合 CPU 密集型任务：

from concurrent.futures import ProcessPoolExecutor
import math

def compute_factorial(n):
    return math.factorial(n)

numbers = [5000, 6000, 7000, 8000]

with ProcessPoolExecutor(max_workers=4) as executor:
    results = executor.map(compute_factorial, numbers)
    
    for n, result in zip(numbers, results):
        print(f"{n}! 有 {len(str(result))} 位数字")

submit vs map

submit：逐个提交

返回 Future 对象，可单独处理每个任务：

from concurrent.futures import ThreadPoolExecutor

def task(n):
    return n * n

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = []
    for i in range(5):
        future = executor.submit(task, i)
        futures.append(future)
    
    # 单独获取每个结果
    for i, future in enumerate(futures):
        result = future.result()  # 阻塞直到完成
        print(f"任务 {i}: {result}")

map：批量处理

自动管理 Future，返回迭代器：

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(task, range(5))
    
    for i, result in enumerate(results):
        print(f"任务 {i}: {result}")

区别：

• submit：更灵活，可以提交不同函数、不同参数
• map：简洁，但所有任务必须调用同一函数

Future 对象

检查状态

from concurrent.futures import ThreadPoolExecutor
import time

def slow_task():
    time.sleep(2)
    return "完成"

with ThreadPoolExecutor() as executor:
    future = executor.submit(slow_task)
    
    print(f"运行中: {future.running()}")
    print(f"已完成: {future.done()}")
    
    time.sleep(1)
    print(f"运行中: {future.running()}")
    
    result = future.result()  # 等待完成
    print(f"已完成: {future.done()}")
    print(f"结果: {result}")

添加回调

from concurrent.futures import ThreadPoolExecutor

def task(n):
    return n * n

def callback(future):
    result = future.result()
    print(f"回调收到结果: {result}")

with ThreadPoolExecutor() as executor:
    future = executor.submit(task, 5)
    future.add_done_callback(callback)
    
    # 继续其他工作
    print("主线程继续")

超时控制

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import time

def slow_task():
    time.sleep(5)
    return "完成"

with ThreadPoolExecutor() as executor:
    future = executor.submit(slow_task)
    
    try:
        result = future.result(timeout=2)  # 最多等待2秒
    except TimeoutError:
        print("任务超时")
        future.cancel()  # 尝试取消（可能已开始执行）

as_completed：按完成顺序处理

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random

def task(name):
    delay = random.uniform(0.5, 2)
    time.sleep(delay)
    return f"{name} ({delay:.2f}秒)"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(task, f"任务{i}"): i for i in range(5)}
    
    # 按完成顺序迭代
    for future in as_completed(futures):
        index = futures[future]
        result = future.result()
        print(f"任务 {index} 完成: {result}")

输出示例（顺序不确定）：

任务 2 完成: 任务2 (0.67秒)
任务 0 完成: 任务0 (0.89秒)
任务 3 完成: 任务3 (1.12秒)
任务 1 完成: 任务1 (1.54秒)
任务 4 完成: 任务4 (1.89秒)

实战案例：批量文件处理

from concurrent.futures import ThreadPoolExecutor, as_completed
import os

def process_file(filepath):
    """模拟文件处理"""
    size = os.path.getsize(filepath)
    # 实际处理...
    return filepath, size

def batch_process(directory, max_workers=4):
    """批量处理目录下的文件"""
    files = [os.path.join(directory, f) 
             for f in os.listdir(directory) 
             if os.path.isfile(os.path.join(directory, f))]
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_file = {executor.submit(process_file, f): f for f in files}
        
        for future in as_completed(future_to_file):
            filepath = future_to_file[future]
            try:
                result_path, size = future.result()
                print(f"处理完成: {result_path} ({size} 字节)")
            except Exception as e:
                print(f"处理失败: {filepath} - {e}")

# 使用示例
# batch_process("/path/to/files")

错误处理

from concurrent.futures import ThreadPoolExecutor

def risky_task(n):
    if n == 3:
        raise ValueError("错误的值")
    return n * n

with ThreadPoolExecutor() as executor:
    futures = [executor.submit(risky_task, i) for i in range(5)]
    
    for i, future in enumerate(futures):
        try:
            result = future.result()
            print(f"任务 {i}: {result}")
        except Exception as e:
            print(f"任务 {i} 失败: {e}")

选择执行器

场景	推荐	原因
网络请求	ThreadPoolExecutor	I/O 阻塞，GIL 无影响
文件读写	ThreadPoolExecutor	I/O 阻塞
数据计算	ProcessPoolExecutor	CPU 密集，绕过 GIL
图像处理	ProcessPoolExecutor	CPU 密集
数据库查询	ThreadPoolExecutor	I/O 阻塞

小结

• concurrent.futures 提供统一的高层并发 API
• ThreadPoolExecutor 管理线程池，适合 I/O 密集型
• ProcessPoolExecutor 管理进程池，适合 CPU 密集型
• submit 逐个提交任务，返回 Future
• map 批量处理，自动管理 Future
• as_completed 按完成顺序迭代结果
• Future 对象提供状态查询、超时控制和回调

回到 并发编程概览 或查看 练习答案。