Python 并行计算实战：优雅封装 ThreadPoolExecutor 提升代码可复用性

在日常的数据处理、日志分析、爬虫采集、风电场 CFD 计算等场景中，经常会遇到任务量大、单任务耗时短的情况。这类任务如果逐个串行执行，效率极低，而借助 Python 标准库 concurrent.futures 提供的 ThreadPoolExecutor，我们可以轻松实现多线程并行计算。

本文将带你从最原始的用法开始，逐步优化封装，最终实现一个 易读、可复用、健壮的并行任务工具函数。

---

1. 最原始的写法

很多人第一次用线程池时，写法大致如下：

from concurrent.futures import ThreadPoolExecutor

def my_task(x):
    return x * x

max_workers = 4
tasks = [1, 2, 3, 4, 5]

with ThreadPoolExecutor(max_workers=max_workers) as ex:
    results = list(ex.map(my_task, tasks))

print(results)  # [1, 4, 9, 16, 25]

这种写法可以快速跑起来，但有几个问题：

1. 耦合度高：my_task、tasks、max_workers 全写死在逻辑里。
2. 缺少异常处理：如果某个任务报错，直接抛异常，影响整个程序。
3. 结果顺序不直观：map 确保顺序，但无法捕捉任务状态。

---

2. 使用 as_completed 捕获结果与异常

为了更灵活地处理任务，可以用 as_completed：

from concurrent.futures import ThreadPoolExecutor, as_completed

def my_task(x):
    if x == 3:
        raise ValueError("测试异常")
    return x * x

tasks = [1, 2, 3, 4, 5]
results = []

with ThreadPoolExecutor(max_workers=3) as ex:
    future_to_task = {ex.submit(my_task, t): t for t in tasks}
    for future in as_completed(future_to_task):
        task = future_to_task[future]
        try:
            result = future.result()
        except Exception as e:
            result = f"任务 {task} 出错: {e}"
        results.append(result)

print(results)
# 可能输出: [1, 4, '任务 3 出错: 测试异常', 16, 25]

优势：

• 每个任务的异常不会导致整体崩溃。
• 可灵活记录日志。

劣势：

• 写法冗长，每次写都要重复模板代码。

---

3. 封装成通用函数：run_in_threads

为避免重复造轮子，我们可以封装成一个函数，让调用更简洁。

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Callable, Iterable, Any

def run_in_threads(
    func: Callable[..., Any],
    tasks: Iterable[tuple],
    max_workers: int = 4,
    show_log: bool = True
) -> list:
    """
    使用线程池并行执行任务。

    参数
    ----------
    func : Callable
        要执行的函数。
    tasks : Iterable[tuple]
        每个任务的参数元组，例如 [(arg1,), (arg1,arg2), ...]。
    max_workers : int, 默认 4
        最大线程数。
    show_log : bool, 默认 True
        是否打印日志。

    返回
    ----------
    results : list
        任务执行结果列表，顺序与传入 tasks 相同。
    """
    results = [None] * len(tasks)
    if show_log:
        print(f"👉 开始并行计算（线程数 = {max_workers}，任务数 = {len(tasks)}）...")

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_idx = {
            executor.submit(func, *args): idx
            for idx, args in enumerate(tasks)
        }
        for future in as_completed(future_to_idx):
            idx = future_to_idx[future]
            try:
                results[idx] = future.result()
            except Exception as e:
                results[idx] = f"❌ 任务 {idx} 出错 → {e}"

    return results

---

4. 使用示例

示例一：数学计算

def add(a, b):
    return a + b

tasks = [(1, 2), (3, 4), (5, 6)]
results = run_in_threads(add, tasks, max_workers=3)
print(results)  # [3, 7, 11]

示例二：模拟 I/O 密集型任务

import time

def fake_io(n):
    time.sleep(1)
    return f"完成任务 {n}"

tasks = [(i,) for i in range(5)]
results = run_in_threads(fake_io, tasks, max_workers=2)
print(results)
# 大约 3 秒完成，而非串行的 5 秒

---

5. 优点总结

• 易读性：逻辑统一封装，调用简洁。
• 可复用性：传入任何函数、参数即可用。
• 健壮性：异常被捕获，不会影响整体任务。
• 顺序一致：输出结果顺序与输入任务顺序完全对应。
• 可扩展：可以加进度条（如 tqdm）、日志记录、结果缓存等。

---

6. 带进度条的增强版本

如果任务较多，可以加上 tqdm 查看实时进度：

from tqdm import tqdm

def run_in_threads_with_progress(func, tasks, max_workers=4):
    results = [None] * len(tasks)
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_idx = {
            executor.submit(func, *args): idx
            for idx, args in enumerate(tasks)
        }
        for future in tqdm(as_completed(future_to_idx), total=len(tasks), desc="进度"):
            idx = future_to_idx[future]
            try:
                results[idx] = future.result()
            except Exception as e:
                results[idx] = f"❌ 任务 {idx} 出错 → {e}"
    return results

运行效果：

进度: 100%|██████████| 50/50 [00:05<00:00,  9.01it/s]

---

7. 适用场景

• 批量计算：风电场机位点风参计算、气象数据处理。
• 日志分析：大规模日志文件的并行解析。
• 网络请求：爬虫或 API 批量调用。
• 数据科学：批量模型训练、文件预处理。

---

8. 总结

本文从最原始的 ThreadPoolExecutor 写法出发，逐步优化到 通用封装函数 run_in_threads，实现了高效、优雅、可复用的并行计算方式。

今后遇到批量计算场景，只需要传入函数和任务参数，几行代码就能轻松并行执行，既能保证健壮性，又能提升开发效率。

🚀 下一步优化方向：

• 加入 任务重试机制（防止偶发错误影响结果）。
• 增加 超时控制（避免某些任务无限卡住）。
• 支持 多进程版本（适合 CPU 密集型任务）。

---

📌 如果你在风能计算、数据分析或爬虫项目里频繁遇到 大批量任务，不妨试试这种写法，能帮你省下不少时间！

---

多进程版（ProcessPoolExecutor） 对比文章，适合 CPU 密集型任务（比如数值模拟、矩阵计算）