Python 异步编程实战：从 asyncio 到结构化并发

背景

在 IO 密集型的 Web 爬虫、批量 PDF 处理、图片压缩场景中，同步代码的性能瓶颈非常明显。本文记录我们把批量图片处理服务从同步重构为异步的完整过程，以及踩到的坑。

同步 vs 异步：什么时候该用 async

# 同步：处理 100 张图片 ≈ 20 秒
for img in images:
    compress(img)   # 每张 0.2s

# 异步（IO 密集）：≈ 3 秒
async def main():
    tasks = [compress_async(img) for img in images]
    await asyncio.gather(*tasks)

关键判断标准：
- CPU 密集（压缩/抠图/OCR 推理）→ 用 multiprocessing 或进程池
- IO 密集（网络请求/文件读写/API 调用）→ 用 asyncio

2024 最佳实践：结构化并发

Python 3.11 引入的 TaskGroup 让结构化并发成为可能：

async def process_batch(file_paths):
    try:
        async with asyncio.TaskGroup() as tg:
            # 批量提交任务
            tasks = [tg.create_task(process_one(p)) for p in file_paths]
            results = [task.result() for task in tasks]
            return results
    except ExceptionGroup as eg:
        # 自动捕获所有子任务异常，不会静默失败
        for exc in eg.exceptions:
            logger.error(f'任务失败: {exc}')

对比传统的 asyncio.gather(..., return_exceptions=True)，TaskGroup 在异常传播、取消语义、生命周期管理上都更清晰。

我们的图片处理服务改造

阶段	方案	吞吐（张/秒）	内存占用
改造前	同步 + 进程池 4 worker	22	稳定
改造中间	asyncio + 进程池	45	略高
最终方案	asyncio + multiprocessing	68	最优

常见陷阱

陷阱 1：同步阻塞调用卡死事件循环

# ❌ 错误
async def handler():
    data = requests.get(url)   # requests 是同步的，会卡事件循环
    return data

# ✅ 正确：用 httpx 或在线程池跑同步代码
async def handler():
    data = await httpx.get(url)
    return data

陷阱 2：过度并发

一张 10MB 的 PDF 合并任务不能开 1000 个协程，磁盘 IO 会成为瓶颈。建议用 asyncio.Semaphore 做限流：

sem = asyncio.Semaphore(20)   # 最多 20 个并发文件处理

async def safe_process(path):
    async with sem:
        return await process_one(path)

总结

异步不是银弹，但在 IO 密集场景下能以极低的成本获得 5-10x 的吞吐提升。关键是：识别瓶颈在哪里、选择正确的并发模型、做合理的限流。