Python asyncio 并发编程指南

理解事件循环、协程与任务，正确处理超时和取消。

核心思路

异步适合大量等待型 I/O，不会自动加速 CPU 密集任务。并发调用用 TaskGroup 组织，所有外部请求设置超时。

落地步骤

取消是正常控制流，清理代码放在 finally。限制并发量，避免瞬间压垮数据库或第三方接口。

验证清单

实施后不要只确认“能运行”，还应检查错误率、延迟、资源占用和异常路径。把验证命令、预期结果和回滚方式写进变更记录，后续维护会轻松很多。

小结

好的工程实践不是堆叠工具，而是用清晰边界、可观察结果和可逆变更降低长期成本。先解决最重要的问题，再逐步自动化。

可执行实践

I/O 并发使用 TaskGroup 或受控 gather，每个外部调用设置 timeout，并用 Semaphore 限制数据库或第三方 API 并发。CPU 密集任务交给进程池或任务队列。

验证与退出条件

正确传播 CancelledError，清理写在 finally。测试超时、取消和部分任务失败，避免创建 task 后不保存引用导致异常无人处理。

实践场景

以“并发调用多个 I/O 服务并可及时取消”为例。实施前先保存当前版本、配置和关键指标基线，写明成功条件与停止条件；涉及生产数据或外部写入时，先在隔离环境验证。

分步实施

协程传递取消语义，TaskGroup 管理子任务；每个外部调用设置超时和并发信号量，阻塞函数移到专用线程。每完成一步都记录可观察结果，不把命令返回零等同于业务正确。

故障演练

模拟一个子任务失败、整体超时和调用方断开，确认其余任务被回收且连接归还池。演练应使用可清理的测试数据，并验证告警、日志和恢复路径，而不只是确认系统最终又能访问。

观测与回滚

观察事件循环延迟、活跃任务和超时率；并发退化时降低上限或回退串行实现。变更单中应写清负责人、观察窗口、阈值和回滚触发器，避免故障发生后再临时争论。

完成后把使用的版本、关键命令、观测窗口和结果写入变更记录。没有验证证据的“已完成”不能作为上线结论。

第二轮实践复核

可复用交付物： 可取消协程、并发上限和故障注入测试。复用时先固定软件版本、依赖锁文件、测试数据规模和运行环境；文章中的命令包含占位符，必须替换成自己的非生产目标后再执行。

最低验证： python -m pytest -q，并检查测试退出后 asyncio.all_tasks() 无残留业务任务

执行记录要保存命令、退出码、关键输出、观察窗口和回滚结果。若验证会产生写入，先使用隔离环境或 dry-run；任何失败都不能通过扩大权限、关闭校验或跳过测试来“解决”。

完整实战：有界并发抓取器

场景是批量访问一组内部只读接口。真正的约束不是“把 1,000 个协程一起启动”，而是总超时、单依赖并发、失败隔离和取消传播。下面示例只使用标准库，适用于支持 TaskGroup 与 timeout 的 Python 版本：

import asyncio
from collections.abc import Awaitable, Callable, Iterable

Fetch = Callable[[str], Awaitable[bytes]]

async def fetch_all(
    urls: Iterable[str],
    fetch: Fetch,
    *,
    concurrency: int = 8,
    timeout_s: float = 5.0,
) -> dict[str, bytes | Exception]:
    semaphore = asyncio.Semaphore(concurrency)
    results: dict[str, bytes | Exception] = {}

    async def one(url: str) -> None:
        try:
            async with semaphore:
                async with asyncio.timeout(timeout_s):
                    results[url] = await fetch(url)
        except asyncio.CancelledError:
            raise  # 取消必须继续向上传播
        except Exception as exc:
            results[url] = exc

    async with asyncio.TaskGroup() as group:
        for url in urls:
            group.create_task(one(url), name=f"fetch:{url}")

    return results

这里选择“单个请求失败不取消整批”，所以异常在 one 内被保存；若业务要求全有或全无，应让异常逃出任务，TaskGroup 会取消同组其他任务。两种语义都合理，关键是明确选择并写测试。

队列比一次性创建全部任务更稳

待处理项可能达到几十万时，即使信号量限制了执行并发，一次性创建全部 Task 仍会占用内存。此时使用固定数量 worker 消费 asyncio.Queue，并为队列设置 maxsize，让生产者在下游变慢时自然背压。

风险	错误做法	可观察的修正
依赖过载	无界 `gather`	信号量/固定 worker，记录在途数
永久挂起	只依赖客户端默认超时	设置单次与整批 deadline
取消失效	捕获所有异常后吞掉取消	单独重新抛出 `CancelledError`
阻塞事件循环	协程里执行 CPU 或同步 I/O	移出主循环或使用受控线程池
重试风暴	每层都自动重试	只在一层按错误类型、预算和抖动重试

必测失败路径

用假 fetch 注入快速成功、超时、连接错误和取消。断言最大在途数不超过预算；取消外层任务后没有遗留任务；超时项被记录且整批能在 deadline 内退出。压测同时观察吞吐、P95 延迟、事件循环延迟、内存峰值和依赖端错误率，不能只比较总耗时。

若部署后依赖错误率或事件循环延迟升高，先把并发配置降回旧值，再停止接收新批次并等待在途任务收敛。并发参数应是可灰度配置，而不是写死在代码中的“最佳值”。

一手资料

资料与版本边界复核日期：2026-08-22。工具行为可能演进，实施前以锁定版本对应的官方文档为准。