Python 并行化技能

将顺序执行的 Python 代码改造为利用并行与并发执行模式。

工作流程

1. 分析代码，识别可并行化的候选部分
2. 分类工作负载类型（CPU 密集型、I/O 密集型或数据并行型）
3. 选择合适的并行化策略
4. 改造代码，加入恰当的同步机制与错误处理
5. 验证正确性，并测量预期的加速比

并行化决策树

瓶颈是 CPU 密集型还是 I/O 密集型？

CPU 密集型（计算密集）：
├── 迭代之间相互独立？ → multiprocessing.Pool / ProcessPoolExecutor
├── 需要共享状态？ → 使用 multiprocessing.Manager 或共享内存
├── 涉及 NumPy/Pandas 操作？ → 优先向量化，再考虑 numba/dask
└── 数据块较大？ → 使用 Pool.map 进行分块处理

I/O 密集型（网络、磁盘、数据库）：
├── 存在大量独立请求？ → 使用 asyncio 配合 aiohttp/aiofiles
├── 基于传统同步代码？ → ThreadPoolExecutor
├── 同步与异步混合？ → asyncio.to_thread()
└── 数据库查询？ → 连接池 + 异步驱动

数据并行型（数组/矩阵运算）：
├── 使用 NumPy 数组？ → 向量化处理，避免 Python 循环
├── 使用 Pandas DataFrame？ → 使用内置向量化方法
├── 处理超大数据集？ → 使用 Dask 实现内存外并行
└── 可用 GPU？ → 考虑 CuPy 或 JAX

改造模式

模式 1：循环 → ProcessPoolExecutor（CPU 密集型）

改造前：

results = []
for item in items:
    results.append(expensive_computation(item))

改造后：

from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor() as executor:
    results = list(executor.map(expensive_computation, items))

模式 2：顺序 I/O → 异步（I/O 密集型）

改造前：

import requests

def fetch_all(urls):
    return [requests.get(url).json() for url in urls]

改造后：

import asyncio
import aiohttp

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def fetch_one(session, url):
    async with session.get(url) as response:
        return await response.json()

模式 3：嵌套循环 → 向量化

改造前：

result = []
for i in range(len(a)):
    row = []
    for j in range(len(b)):
        row.append(a[i] * b[j])
    result.append(row)

改造后：

import numpy as np
result = np.outer(a, b)

模式 4：CPU 与 I/O 混合任务（使用 asyncio）

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def hybrid_pipeline(data, urls):
    loop = asyncio.get_event_loop()

    # 在进程池中执行 CPU 密集型任务
    with ProcessPoolExecutor() as pool:
        processed = await loop.run_in_executor(pool, cpu_heavy_fn, data)

    # 异步执行 I/O 密集型任务
    results = await asyncio.gather(*[fetch(url) for url in urls])

    return processed, results

并行化候选模式

在代码中查找以下模式：

安全性要求

并行化时必须始终保障程序正确性：

1. 识别共享状态 —— 跨迭代被修改的变量会破坏并行性
2. 检查依赖关系 —— 若第 N 次迭代依赖第 N−1 次结果，则必须顺序执行
3. 异常处理 —— 将并行代码包裹在 try/except 中；如需细粒度错误控制，请使用 executor.submit()
4. 资源管理 —— 使用上下文管理器；限制工作线程/进程数量，防止资源耗尽
5. 保持顺序 —— 当结果顺序至关重要时，应使用 map() 而非 submit()

常见陷阱

• GIL 陷阱：线程对 CPU 密集型 Python 代码无效——应使用 multiprocessing
• Pickle 失败：lambda 函数与嵌套类无法被序列化用于 multiprocessing
• 内存爆炸：ProcessPoolExecutor 会将数据拷贝至每个子进程——对大数据请使用共享内存
• 同步代码强行异步化：不能仅添加 async 关键字——需重构整个调用链
• 过度并行化：小规模任务（通常少于 1000 项）的并行开销可能超过收益

验证清单

在最终确认改造后的代码前，请完成以下检查：

• 对测试输入，输出与原始顺序版本一致
• 不存在竞态条件（共享可变状态已正确同步）
• 异常已被捕获并恰当处理
• 资源已正确释放（线程池已关闭、连接已释放等）
• 工作线程/进程数量已限定（使用默认值或显式设置上限）
• 已添加所有必需的 import 语句