139 lines
5.8 KiB
Markdown
139 lines
5.8 KiB
Markdown
---
|
||
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 的区别、怎么限制并发、超时和取消如何实现、怎么在异步代码中调用同步库。
|