Python asyncio 并发实战：用 Semaphore 控制并发，用 TaskGroup 管理任务组

写在前面

写接口服务时，你大概率遇到过这样的情况：上游有 200 个 URL 要抓，下游要同时请求 10 个数据库。随手写个 for 循环串行跑，耗时 40 分钟；改成线程池，速度是快了，可 CPU 占用飙到 90%，下游数据库连接池直接被打满，连接池告警把值班同学半夜叫醒。

Python 的 asyncio 正好解决这个中间地带：用一个线程就能同时推进成千上万个 I/O 等待，并且给你精确的并发控制旋钮。这篇文章用一个小例子把两个最关键的旋钮讲清楚——Semaphore（限并发）和 TaskGroup（管任务组）。

一、最小可运行示例

先看一个能直接跑起来的版本。我们模拟去 5 个站点抓取首页标题，每个请求耗时用 asyncio.sleep 模拟 1 秒。

import asyncio
import time

async def fetch(site: str) -> str:
    await asyncio.sleep(1)          # 模拟一次网络 I/O
    return f"{site} OK"

async def main():
    sites = [f"site-{i}" for i in range(5)]

    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(s) for s in sites))
    cost = time.perf_counter() - start

    print(results)
    print(f"总耗时: {cost:.2f}s")

asyncio.run(main())

串行执行需要 5 秒，而这段代码只用了约 1 秒。原因很简单：5 个 await asyncio.sleep(1) 被挂起时，事件循环立刻去跑下一个，整个过程只在一个线程里交替推进。

asyncio.gather 在这里扮演的角色是"把一堆 coroutine 打包成一个可 await 的整体"。但它有个明显缺陷：任何一个子任务抛异常，整个 gather 立刻抛异常，其余任务不会被取消，而是继续在后台跑着。这在生产环境里很容易造成"日志报错但请求还在发"的诡异现象。

二、用 Semaphore 把并发闸门拧到合适的位置

并发不是越高越好。下游服务通常有自己的承载上限，我们必须在客户端侧就设一道闸门，这正是 asyncio.Semaphore 的职责。

import asyncio
import time

MAX_CONCURRENCY = 2      # 最多同时跑 2 个请求

async def fetch(site: str) -> str:
    await asyncio.sleep(1)
    return f"{site} OK"

async def fetch_limited(site: str, sem: asyncio.Semaphore) -> str:
    async with sem:                      # 拿不到许可就在这里排队等待
        return await fetch(site)

async def main():
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    sites = [f"site-{i}" for i in range(5)]

    start = time.perf_counter()
    results = await asyncio.gather(*(fetch_limited(s, sem) for s in sites))
    cost = time.perf_counter() - start

    print(results)
    print(f"总耗时: {cost:.2f}s")   # 约 3s：5 个任务，闸门开 2

asyncio.run(main())

实测下来总耗时约 3 秒，正好是 ceil(5/2) × 1s。三个要点值得记住：

1. async with sem: 包裹的是真正的耗时部分，不要把整个函数都包进去，否则等于没限流。
2. 信号量的许可对象是可复用的，一次 async with 结束时许可会自动归还，不需要手动 release()。
3. 信号量在创建时绑定事件循环，所以要写成 asyncio.Semaphore() 这种工厂式调用（Python 3.10+ 移除了循环参数），不要在模块顶层创建全局实例。

如果你的并发数需要动态调整（比如根据 CPU 核数或响应延迟自适应），Semaphore 本身不支持增减，但可以自己包一层：一个带 asyncio.Condition 的简易令牌池。不过大多数场景下，先从"固定数值 + 压测"开始就够了。

三、用 TaskGroup 替代 gather 管理任务组

Python 3.11 引入的 asyncio.TaskGroup 正好补上了 gather 的短板。它是一个上下文管理器，进入时会创建一个任务组，退出时会等待组内所有任务完成；如果任何一个任务失败，剩余任务会被立即取消，并抛出 ExceptionGroup。

import asyncio

async def fetch(site: str) -> str:
    await asyncio.sleep(1)
    if site == "site-2":
        raise RuntimeError(f"{site} 挂了")
    return f"{site} OK"

async def main():
    sites = [f"site-{i}" for i in range(5)]

    async with asyncio.TaskGroup() as tg:          # 3.11+
        tasks = [tg.create_task(fetch(s)) for s in sites]

    print([t.result() for t in tasks])

asyncio.run(main())

输出是一个 ExceptionGroup，包含 site-2 的 RuntimeError，同时其他任务被取消，程序不会留下"孤儿任务"在后台继续跑。这是做批量任务时非常关键的一个保证：要么全成功，要么明确失败，不会出现半死不活的状态。

需要注意两个细节：

• 异常被包装成了 ExceptionGroup，捕获时要用 except* 语法（Python 3.11+），否则会漏掉异常。

  try:
      async with asyncio.TaskGroup() as tg:
          ...
  except* RuntimeError as eg:
      print(eg.exceptions)
  

• 不支持 Python 3.10 及以下。如果生产环境版本偏老，可以退回到 gather + return_exceptions=True 的组合，但要自己处理部分成功的语义。

另外，TaskGroup 内部还可以用 tg.create_task() 动态追加任务，任务组会跟踪所有由它创建的任务，退出上下文时统一等待。这对"边遍历边并发"这类模式非常友好。

四、超时、取消与优雅退出

线上服务最怕的就是一个请求卡死把整个协程池拖住。给每个任务加上超时是必做项：

import asyncio

async def fetch(site: str) -> str:
    await asyncio.sleep(1)
    return f"{site} OK"

async def fetch_with_timeout(site: str, timeout: float = 3.0) -> str:
    try:
        return await asyncio.wait_for(fetch(site), timeout=timeout)
    except asyncio.TimeoutError:
        return f"{site} TIMEOUT"

async def main():
    sites = [f"site-{i}" for i in range(5)]
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_with_timeout(s)) for s in sites]
    print([t.result() for t in tasks])

asyncio.run(main())

asyncio.wait_for 在超时时抛出 TimeoutError，同时会取消内部的协程。配合 TaskGroup 使用时，取消会沿任务树向上传播，整个任务组会一并退出——这正是我们想要的"要么全完成要么整体回滚"的语义。

如果是主动取消（收到 SIGTERM 需要下线），CancelledError 在 Python 3.8 之后属于 BaseException 而非 Exception，所以不要用 except Exception 吞掉它，否则会破坏取消语义：

try:
    await do_work()
except asyncio.CancelledError:
    await cleanup()      # 做清理，然后继续向上抛
    raise
except Exception:
    logger.exception("普通异常")

五、什么时候不该用 asyncio

asyncio 不是银弹。下面几种情况，线程池甚至同步代码反而更合适：

• CPU 密集型任务（图像处理、大量数值计算、压缩解压）。asyncio 对 CPU 密集型没有任何帮助，因为 GIL 的存在让协程依然无法并行执行。这种场景应该用 ProcessPoolExecutor 或直接上 multiprocessing。
• 已经在用同步库且没有异步版本。把同步调用丢进 asyncio.to_thread() 是可以的，但要注意线程池默认上限（min(32, cpu_count+4)），盲目并发反而会因为线程竞争而变慢。
• 调用方数量很少、总量很小。为了一次请求去搭异步框架，收益抵不上复杂度成本。

六、小结

今天聊了三个点：

1. asyncio.gather 适合"全成功或全失败"的简单场景，但要注意它不会取消其他任务。
2. asyncio.Semaphore 是控制并发量最直接的手段，把它包在真正的耗时逻辑外面。
3. Python 3.11+ 优先用 asyncio.TaskGroup，失败即取消，异常统一用 except* 捕获。

把这三者组合起来，就是一个既能撑住并发量、又能干净失败退出的批量任务骨架。如果你的服务里正好有这类"要并发、要限流、要可控失败"的场景，不妨照着这个骨架改一改。

欢迎在评论区交流你在实际项目里遇到的并发坑，我看到会回复。