AI Agent Engineer

AI Agent Engineer

异步、并发、取消与限流

7 min read

异步、并发、取消与限流

本节目标

理解协程为何适合大量等待型 I/O，能用结构化并发管理任务生命周期，并为超时、取消、并发上限和部分失败设计明确行为。这里不把并发当作默认优化，也不把它误称为 CPU 并行。

先判断是否需要并发

工作类型主要时间花在哪里常见选择
I/O-bound等网络、磁盘、数据库、模型服务asyncio，或受控线程
CPU-bound编码、图像处理、大量 Python 计算测量后考虑进程、原生库或其他运行时
很小或必须顺序数据有依赖、调用量低普通同步代码更清楚

并发（concurrency）表示多个任务的执行区间重叠；并行（parallelism）表示同一时刻真的在多个计算资源上运行。单线程事件循环可以高效并发等待，但不会自动让 CPU 密集 Python 代码更快。

协程与事件循环

调用 async def 函数只创建协程对象；只有被 await 或注册为任务后才执行：

import asyncio
 
 
async def fetch_one(name: str) -> str:
    await asyncio.sleep(0.05)  # 模拟非阻塞等待
    return name.upper()
 
 
async def main() -> None:
    result = await fetch_one("rag")
    print(result)
 
 
if __name__ == "__main__":
    asyncio.run(main())

在库代码中通常不调用 asyncio.run()；它属于应用最外层入口。Notebook、Web 框架和异步服务已经拥有事件循环，重复启动会报错。

用 TaskGroup 管理生命周期

Python 3.11+ 的 asyncio.TaskGroup 将“创建哪些任务、何时等待、一个失败后其余任务怎么办”放进同一作用域：

async def fetch_all(names: list[str]) -> list[str]:
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch_one(name)) for name in names]
    return [task.result() for task in tasks]

当某个子任务首次抛出非取消异常时，TaskGroup 会取消其他尚未完成的任务；退出时可能用 ExceptionGroup 聚合多个异常。它适合“这些子任务共同构成一次操作”的 fail-fast 语义。

若业务允许部分成功，应显式把每项结果建模为成功/失败记录，而不是用宽泛 gather(..., return_exceptions=True) 后忘记检查异常。

并发上限与背压

一次创建十万个请求即使都异步，也会耗尽连接、内存或服务配额。Semaphore 限制同时进入外部边界的数量：

async def bounded_map(names: list[str], limit: int) -> list[str]:
    if limit < 1:
        raise ValueError("limit 必须至少为 1")
    semaphore = asyncio.Semaphore(limit)
 
    async def run(name: str) -> str:
        async with semaphore:
            return await fetch_one(name)
 
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(run(name)) for name in names]
    return [task.result() for task in tasks]

并发上限不等于每秒速率限制。前者限制同时在途数量，后者限制时间窗口内请求数；生产客户端往往同时需要服务端配额、队列长度和总任务预算。

超时与取消是控制流

async def fetch_with_deadline(name: str) -> str:
    try:
        async with asyncio.timeout(2.0):
            return await fetch_one(name)
    except TimeoutError as exc:
        raise TemporaryDependencyFailure("读取超时") from exc

超时作用域取消当前任务，并在作用域外转换为内置 TimeoutError。取消可能在任意 await 点到达；用 try/finally 收尾，然后重新抛出 CancelledError：

async def worker() -> None:
    resource = await open_resource()
    try:
        await use_resource(resource)
    except asyncio.CancelledError:
        # 只做必要的取消记录或状态转换
        raise
    finally:
        await resource.aclose()

吞掉 CancelledError 会破坏 TaskGroup、超时和应用关闭。除非真的要把取消转成一种已定义业务结果，否则必须传播。

不要阻塞事件循环

以下调用会阻塞当前线程，不能直接放在异步热路径：

  • time.sleep()；
  • 同步 HTTP/数据库 SDK；
  • 大文件同步读写；
  • CPU 密集循环。

小型不可避免的同步 I/O 可用 await asyncio.to_thread(function, *args) 隔离，但线程中的函数仍可能在协程取消后继续运行。必须设置底层超时并控制提交量；不要把 to_thread 当作可强制终止任意代码的机制。

结果顺序、重复与共享状态

  • 完成顺序不一定等于输入顺序；接口应明确使用哪一种；
  • 外部调用超时后可能已成功，重试仍需幂等语义；
  • 避免多个协程直接修改同一可变字典；优先让任务返回值，再在单一位置汇总；
  • 若必须共享状态，使用锁并把临界区保持很小，且不可在持锁时等待慢 I/O。

练习

  1. 用 asyncio.sleep() 模拟 20 个任务，将并发上限设为 3；记录并证明峰值不超过 3。
  2. 让第 5 项失败，比较 TaskGroup 的 fail-fast 与“每项返回结果对象”的部分成功设计。
  3. 给整个批次设置 0.3 秒总时限，检查所有任务是否收尾。
  4. 把 time.sleep() 错放进协程并测量，再改为 asyncio.sleep()；解释差异。

自测

  • 能区分协程对象、任务和事件循环。
  • 能解释并发、并行、并发上限和速率限制的差别。
  • 能说明 TaskGroup 中一个任务失败后发生什么。
  • 能正确传播取消，并指出资源关闭路径。
  • 能识别会阻塞事件循环的同步调用。

相关概念与下一步

参考资料

获取日期：2026-07-14。课程以 Python 3.11+ 的稳定接口为基线，不依赖 3.14 新增的 TaskGroup.create_task() 参数。

Backlinks