---
id: python.asyncio.concurrency-control
title: 并发控制：gather / TaskGroup、Semaphore、超时、取消、阻塞调用
type: lesson
domain: python
stack: [asyncio]
tags: [TaskGroup, Semaphore, timeout, 取消, to_thread, 背压]
level: intermediate
prerequisites: [python.asyncio.coroutines-tasks]
est_hours: 3
related: [python.asyncio.scenario.concurrent-llm-calls, python.asyncio.interview, backend.go.errors-concurrency]
week: [4, 5, 12]
order: 2
---

# 并发控制：gather / TaskGroup、Semaphore、超时、取消、阻塞调用

## 解释说明

**等待一组任务的三种方式**

| 写法 | 一个失败时 | 适合 |
| --- | --- | --- |
| `gather(*coros)` | 第一个异常立即抛给调用方，**其他任务继续运行**（不会被取消） | 简单的并发 |
| `gather(*coros, return_exceptions=True)` | 异常作为结果返回，不抛出 | 允许部分失败 |
| `async with asyncio.TaskGroup() as tg:`（3.11+） | **取消其余任务**，异常合并为 `ExceptionGroup` 抛出 | 要么全部成功、要么全部取消 |

**限制并发**：`asyncio.Semaphore(n)`，`async with sem:` 包住真正的调用。没有限制时，一次性并发 1000 个请求会触发下游限流、耗尽连接池。（对照 Go：带缓冲的 channel 当信号量，或 `errgroup.SetLimit`。）

**超时**

- `async with asyncio.timeout(秒):`（3.11+）或 `await asyncio.wait_for(coro, 秒)`；超时抛 `TimeoutError`，内部的任务会被取消。
- 分清**单次调用超时**和**整体截止时间**（例如接口必须 10 秒内返回，内部每次模型调用最多 5 秒）。

**取消**

- `task.cancel()` 会在任务下一个 `await` 点抛出 `CancelledError`。
- 捕获 `CancelledError` 做清理后**必须重新抛出**，否则取消被吞掉，调用方以为任务还在正常运行。
- `except Exception` 不会捕获 `CancelledError`（它继承自 `BaseException`），这是有意设计的。

**阻塞调用**：必须用同步库（某些 SDK、文件解析、CPU 计算）时，`await asyncio.to_thread(func, 参数)` 放到线程池执行，不阻塞事件循环。CPU 密集且耗时长的用进程池。

**背压**：生产者比消费者快时，用有界的 `asyncio.Queue(maxsize=n)`，队列满时 `put` 会等待。

## 案例

批量为 2000 篇文档生成摘要：用 Semaphore 限制最多 8 个并发请求（符合模型服务商的速率限制）；每次调用 30 秒超时；限流错误（429）指数退避重试；单篇失败记录下来，最后汇总重跑，而不是让整个批次失败。

## 代码示例

```python
import asyncio


async def fetch(i: int) -> int:
    await asyncio.sleep(0.05 * (i % 3))
    if i == 4:
        raise ValueError(f"任务 {i} 失败")
    return i * 10


async def main() -> None:
    # 1. gather + return_exceptions：部分失败也拿到所有结果
    results = await asyncio.gather(*(fetch(i) for i in range(6)), return_exceptions=True)
    print([r if not isinstance(r, Exception) else f"错误：{r}" for r in results])

    # 2. TaskGroup：一个失败，其余被取消，异常以 ExceptionGroup 抛出
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(i)) for i in range(6)]
    except* ValueError as eg:                       # except* 处理 ExceptionGroup（3.11+）
        print("TaskGroup 失败：", [str(e) for e in eg.exceptions])
    print("被取消的任务数：", sum(t.cancelled() for t in tasks))

    # 3. Semaphore 限制并发
    sem = asyncio.Semaphore(2)
    running = 0

    async def limited(i: int) -> None:
        nonlocal running
        async with sem:
            running += 1
            assert running <= 2
            await asyncio.sleep(0.01)
            running -= 1

    await asyncio.gather(*(limited(i) for i in range(10)))
    print("Semaphore：同时运行的任务从未超过 2 个")

    # 4. 超时
    try:
        async with asyncio.timeout(0.1):
            await asyncio.sleep(1)
    except TimeoutError:
        print("超时")

    # 5. 取消：清理后重新抛出
    async def worker() -> None:
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("收到取消，释放资源")
            raise                                   # 必须重新抛出

    t = asyncio.create_task(worker())
    await asyncio.sleep(0.01)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("任务已取消：", t.cancelled())

    # 6. 同步阻塞函数放到线程里执行
    def parse_pdf_sync(path: str) -> str:
        import time
        time.sleep(0.1)                             # 模拟同步库的阻塞调用
        return f"{path} 的文本"

    print(await asyncio.to_thread(parse_pdf_sync, "manual.pdf"))


asyncio.run(main())
```

把这些组合起来（并发上限 + 单次超时 + 重试 + 部分失败 + 整体截止时间）的完整实现和测试见 [examples/fanout](examples/fanout/)。

## 易错点

- **用 `gather` 以为失败时会取消其他任务**：不会，其他任务继续跑（可能继续消耗模型额度）。需要“一个失败全部取消”用 TaskGroup。
- **吞掉 `CancelledError`**：任务无法被取消，服务关闭时卡住。
- **没有并发上限**：一次性创建成千上万个请求，下游限流、连接池耗尽、内存上涨。
- **只设整体超时不设单次超时**（或反过来）：一个慢请求占满全部时间，或者总时长失控。
- **重试不区分错误类型**：参数错误（400）重试多少次都一样失败，只重试超时、429、5xx。
- **在 `to_thread` 里跑 CPU 密集任务**：受 GIL 限制，没有并行效果，用进程池。

## 面试常问

见 [asyncio 面试题](interview.md)：gather 和 TaskGroup 的区别、怎么限制并发、超时和取消如何实现、怎么在异步代码中调用同步库。