金猪博客 金猪博客

把 asyncio 写到能停下来：TaskGroup、取消、超时与背压

Chen Xi
Chen Xi
  2026-08-05 22:20:00   2026-08-05 22:20:00  

很多 Python 异步教程最后都会停在这里：同时创建几个协程，然后 gather 一下，看着总耗时从 5 秒变成 1 秒。

这当然没错，但它只回答了“任务能不能同时跑”，没有回答更难的问题：任务出错时怎么办？某个请求超时以后，其他任务要不要停？用户按下 Ctrl+C，程序能不能把手里的资源收拾干净？生产者比消费者快很多时，内存会不会一路涨上去？

异步程序真正难的部分不是 async 和 await，而是生命周期。

这篇文章从一个看似合理的 gather() 版本开始，逐步改造成一个可以控制的任务处理器。整个示例只使用 Python 标准库，目标不是实现一个完整爬虫，而是把并发系统中最容易被忽略的边界讲清楚。

image

版本范围

文中的基本写法适用于 Python 3.11 及以上版本：

  • asyncio.TaskGroup 在 Python 3.11 加入；
  • asyncio.timeout() 在 Python 3.11 加入；
  • ExceptionGroup 和 except* 同样来自 Python 3.11；
  • 示例最后使用的 asyncio.Queue.shutdown() 需要 Python 3.13 及以上。

如果项目仍然运行在 Python 3.11 或 3.12，可以把队列关闭部分替换为哨兵值，任务管理的核心思想不变。当前 Python 文档已经把 TaskGroup、取消传播和超时放在结构化并发的同一组 API 中；这不是偶然的，因为超时和任务组本质上都依赖取消机制。

一个能跑，但不够可靠的版本

先写一个最常见的并发处理器。这里用睡眠模拟 I/O，不访问真实网站，避免把网络不稳定误认为并发问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import asyncio


async def handle(item: str) -> str:
    await asyncio.sleep(0.2)
    if item == "bad":
        raise RuntimeError("处理失败")
    return f"done:{item}"


async def main() -> None:
    items = ["a", "b", "bad", "c"]
    tasks = [asyncio.create_task(handle(item)) for item in items]
    results = await asyncio.gather(*tasks)
    print(results)


asyncio.run(main())

这个程序可以并发运行，但它有几个工程问题。

第一，bad 抛出异常后，gather() 默认会把异常交给调用方。其他任务是否已经完成、是否还在后台运行，需要调用方自己继续确认。第二，任务没有名字，出现问题时日志很难对应到输入。第三，没有超时，没有重试，也没有限制同时运行的数量。把 items 从 4 个换成 400 万个，程序会一次性创建 400 万个任务。

gather() 适合简单的“把几件独立的事一起等完”。一旦任务拥有共同的生命周期，就需要更明确的结构。

先理解取消：它是请求，不是按钮

调用 task.cancel() 并不意味着任务已经立刻结束。它只是向任务发出取消请求；任务下一次在可中断的 await 处恢复时，才会收到 asyncio.CancelledError。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import asyncio


async def worker() -> None:
    try:
        await asyncio.sleep(60)
    except asyncio.CancelledError:
        print("收到取消请求，释放资源")
        raise
    finally:
        print("无论成功、失败还是取消，都会执行清理")


async def main() -> None:
    task = asyncio.create_task(worker(), name="demo-worker")
    await asyncio.sleep(0.1)
    task.cancel("shutdown")
    try:
        await task
    except asyncio.CancelledError:
        print("主程序确认任务已经取消")


asyncio.run(main())

这里有三个细节值得记住。

CancelledError 直接继承自 BaseException，不是普通的 Exception。一般的 except Exception 不会把它吞掉，这是好事；如果代码写了 except BaseException，就必须非常小心。

如果确实需要捕获取消异常，清理完成以后通常应该再次 raise。一个任务如果把取消吞掉，外层的超时和 TaskGroup 可能无法得到正确状态。Python 文档也特别提醒，结构化并发组件内部使用取消来协调任务；随意吞掉 CancelledError 会使它们行为异常。

TaskGroup：把任务放进同一个生命周期

TaskGroup 的核心不是少写几行代码，而是表达一个关系：这一组任务属于同一个工作单元。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import asyncio


async def handle(item: str) -> str:
    await asyncio.sleep(0.2)
    if item == "bad":
        raise RuntimeError(f"{item} 失败")
    return f"done:{item}"


async def main() -> None:
    async with asyncio.TaskGroup() as group:
        for item in ["a", "b", "bad", "c"]:
            group.create_task(handle(item), name=f"handle:{item}")


try:
    asyncio.run(main())
except* RuntimeError as errors:
    for error in errors.exceptions:
        print(f"记录任务错误：{error}")

当组内某个任务抛出非取消异常时，TaskGroup 会取消其他仍在运行的任务，等待它们完成清理，再把错误组合成 ExceptionGroup 抛出。这样就不会出现“主程序已经返回，某些任务还偷偷留在后台”的模糊状态。

这也是它和 gather() 的思路差异：

场景 gather() TaskGroup
几个互不相关的结果 简洁 也可以使用，但表达力稍重
一组任务共同成功或失败 需要自己编排 默认就是这个语义
一个任务失败后取消兄弟任务 需要额外处理 内置处理
等待所有任务清理完毕 容易遗漏 上下文退出时保证等待
失败信息 单个异常或返回值 ExceptionGroup

TaskGroup 并不会替你决定业务策略。比如“单个 URL 失败是否应该让整批任务失败”，仍然需要在 worker 内部捕获并记录；但它至少会让未处理的异常沿着明确的边界传播。

超时应该包住哪一段

超时不是给任务贴一个数字，而是给一段等待设置截止时间。现代写法是 asyncio.timeout()：

1
2
3
4
5
6
async def fetch_one(item: str, limit: float) -> str:
    try:
        async with asyncio.timeout(limit):
            return await handle(item)
    except TimeoutError:
        return f"timeout:{item}"

注意 TimeoutError 应该在上下文管理器外面捕获。asyncio.timeout() 在截止时间到达时取消当前任务，并在上下文退出时把内部的 CancelledError 转换为 TimeoutError。如果在上下文内部捕获，通常捕获不到最终的 TimeoutError。

wait_for() 仍然有用，尤其是只需要等待一个 awaitable 时：

1
result = await asyncio.wait_for(handle("one"), timeout=0.5)

但在一段代码中需要嵌套多个 await、动态调整截止时间或和 TaskGroup 配合时，timeout() 更容易表达“这一段工作的总预算”。

还有一个容易误解的点：超时不等于底层工作已经瞬间消失。取消仍然需要在任务的下一个可中断位置发生，finally 清理也需要时间。所以一个严格的上层超时，不应该假设所有底层系统都能在同一纳秒停止。

有界队列：背压是稳定性功能

并发数和队列长度解决的是两个不同的问题。

  • 并发数限制同时处理多少任务；
  • 队列长度限制等待中的任务最多积累多少。

如果生产者不停地读取数据，消费者却因为网络或磁盘变慢，无界队列会把压力转化成内存增长。asyncio.Queue(maxsize=N) 在队列满时让 put() 等待，这就是背压：下游变慢，上游也必须慢下来。

1
2
3
4
5
6
7
queue: asyncio.Queue[str] = asyncio.Queue(maxsize=100)


async def producer(queue: asyncio.Queue[str], items: list[str]) -> None:
    for item in items:
        await queue.put(item)
        print(f"生产 {item}，队列长度={queue.qsize()}")

maxsize=100 不是“优化参数”，而是系统的内存边界之一。它越大，吞吐可能更平滑，但故障时积压更多；它越小，背压来得更早，但生产端更容易等待。真正合适的值要结合任务对象大小、单个任务耗时和可接受的排队延迟测量，而不是照搬别人的数字。

一个完整的可控任务处理器

下面把几个概念组合起来。这个版本使用 Python 3.13 的 Queue.shutdown()，关闭生产端后，消费者会把队列里已经放入的项目处理完，再收到 QueueShutDown 退出。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
from __future__ import annotations

import asyncio
from dataclasses import dataclass
from random import Random


@dataclass(frozen=True, slots=True)
class Job:
    number: int
    name: str


@dataclass(frozen=True, slots=True)
class Outcome:
    job: Job
    value: str | None = None
    error: str | None = None


async def do_io(job: Job, rng: Random) -> str:
    """模拟一个不稳定的 I/O 操作。"""
    delay = rng.uniform(0.05, 0.45)
    await asyncio.sleep(delay)
    if job.name == "bad":
        raise ValueError("服务返回了不可用数据")
    return f"value-for-{job.name}"


async def run_one(job: Job, rng: Random) -> Outcome:
    try:
        async with asyncio.timeout(0.30):
            value = await do_io(job, rng)
        return Outcome(job=job, value=value)
    except TimeoutError:
        return Outcome(job=job, error="timeout")
    except ValueError as error:
        return Outcome(job=job, error=str(error))


async def produce(
    queue: asyncio.Queue[Job],
    jobs: list[Job],
) -> None:
    try:
        for job in jobs:
            await queue.put(job)
    finally:
        # 默认 shutdown 会允许消费者把已经入队的项目处理完。
        queue.shutdown()


async def worker(
    name: str,
    queue: asyncio.Queue[Job],
    results: list[Outcome],
) -> None:
    # 每个 worker 使用自己的随机源；真实项目中也可以注入客户端或连接池。
    rng = Random(name)
    while True:
        try:
            job = await queue.get()
        except asyncio.QueueShutDown:
            return

        try:
            result = await run_one(job, rng)
            results.append(result)
        except asyncio.CancelledError:
            # 被外部停止时，不要吞掉取消；finally 仍然负责标记本次取出的任务。
            raise
        finally:
            queue.task_done()


async def process(jobs: list[Job], worker_count: int = 3) -> list[Outcome]:
    queue: asyncio.Queue[Job] = asyncio.Queue(maxsize=4)
    results: list[Outcome] = []

    async with asyncio.TaskGroup() as group:
        producer_task = group.create_task(produce(queue, jobs), name="producer")
        for index in range(worker_count):
            group.create_task(
                worker(f"worker-{index}", queue, results),
                name=f"worker-{index}",
            )

        # 让生产者的异常进入 TaskGroup，而不是在后台丢失。
        await producer_task
        await queue.join()

    return sorted(results, key=lambda result: result.job.number)


async def main() -> None:
    jobs = [
        Job(1, "alpha"),
        Job(2, "beta"),
        Job(3, "bad"),
        Job(4, "gamma"),
        Job(5, "delta"),
    ]
    outcomes = await process(jobs)
    for outcome in outcomes:
        print(outcome)


asyncio.run(main())

这段代码里每个组件都有单一职责：produce() 只负责投递，worker() 只负责消费，run_one() 只负责一次带超时的业务操作，process() 负责把它们放进同一个生命周期。

还有三个值得注意的决定。

业务错误不一定要取消整组任务

run_one() 把预期中的超时和数据错误转成 Outcome。这样单个工作项失败不会让整批任务失效，最后可以统一统计成功与失败。

但如果连接池损坏、配置缺失或程序出现真正的编程错误，就不应该把异常全部转成字符串。让它穿过 worker() 进入 TaskGroup，兄弟任务会被取消，程序也会得到明确的失败信号。

换句话说，要区分：

1
2
业务失败：这个项目失败，但系统还能继续
系统失败：继续运行可能只会制造更多错误

task_done() 必须与 get() 配对

queue.join() 等待的是“每一次 put() 都对应一次 task_done()”。如果 worker 在 get() 之后直接抛出异常，没有进入 finally，主任务可能永远卡在 join()。

这也是为什么 task_done() 放在 finally 里。这里的语义是：任务已经从队列取出，本次处理生命周期结束；如果业务需要在取消时重新入队，则应该显式设计重试队列，而不是偷偷遗漏一次 task_done()。

正常关闭和立即停止是两种不同语义

1
queue.shutdown()

表示不再接受新项目，但允许已经入队的项目排空。适合正常收尾。

1
queue.shutdown(immediate=True)

表示尽快打断队列处理，已经排队的项目可能不会执行。适合进程即将被强制终止，或者任务已经失去继续处理的价值。它会破坏“每次 put 都最终 task_done”的直觉，因此不应该当作普通停止按钮。

如果使用 Python 3.12 或更早版本，可以用唯一的哨兵对象通知 worker 退出：生产者完成后向队列放入与 worker 数量相同的哨兵。哨兵方案能工作，但需要自己保证数量、顺序和取消路径；3.13 的 shutdown() 把这部分状态收进了队列对象。

外部取消如何传到最里面

实际程序通常还有一层服务入口：定时任务、命令行程序或 Web 请求会启动 process()。外部停止时，取消应该沿着调用栈向下传播。

1
2
3
4
5
6
async def serve() -> None:
    try:
        await process(make_jobs())
    except asyncio.CancelledError:
        print("服务正在停止")
        raise

不要在最外层写一个无条件的：

1
2
except BaseException:
    return

它会同时吞掉 KeyboardInterrupt、系统退出信号和任务取消。程序表面上“没有报错”，实际上可能留下半开的连接、未刷新的文件和未提交的结果。

更可靠的关闭顺序通常是：

1
2
3
4
5
停止接收新任务
→ 让队列进入关闭状态
→ 等待正在运行的任务清理
→ 关闭客户端、连接池和文件
→ 把取消异常继续交给上层

取消是控制流的一部分，不是异常日志里的噪音。

重试不应该和取消混在一起

网络任务常常需要重试，但“请求失败”和“上层要求停止”不是一回事。重试装饰器如果捕获了所有异常，就可能在程序已经关闭时继续睡眠和重试。

一个简单的重试循环应该把取消放在明确的分支里：

1
2
3
4
5
6
7
8
9
10
11
12
async def retrying(job: Job, attempts: int = 3) -> Outcome:
    for attempt in range(1, attempts + 1):
        try:
            async with asyncio.timeout(0.3):
                value = await do_io(job, Random(attempt))
            return Outcome(job=job, value=value)
        except asyncio.CancelledError:
            raise
        except (TimeoutError, OSError) as error:
            if attempt == attempts:
                return Outcome(job=job, error=f"failed after {attempts}: {error}")
            await asyncio.sleep(0.05 * 2 ** (attempt - 1))

重试次数、退避时间和可重试异常应该是业务策略，不应由并发框架偷偷决定。对于不可重试的参数错误，应该立即失败；对于取消，应该立即传播。

用指标观察“可控”是否真的成立

只看总耗时，无法知道一个异步系统是否健康。至少应该记录：

  • 队列峰值长度；
  • 当前运行任务数；
  • 每个任务等待时间和处理时间；
  • 超时、重试、业务失败和系统失败次数；
  • 取消发生时仍有多少任务在运行；
  • 关闭后是否还有未完成队列项。

例如，可以在 put() 前后记录队列长度，在 run_one() 周围记录单调时钟：

1
2
3
4
loop = asyncio.get_running_loop()
started = loop.time()
result = await run_one(job, rng)
elapsed = loop.time() - started

不要用 time.time() 计算耗时，因为系统时钟可能被校准或回拨；事件循环提供的单调时钟才适合测量间隔。

gather() 什么时候仍然合适

写完 TaskGroup 后，不需要把 gather() 视为过时 API。它仍然适合以下场景：

  • 一次并行读取几份互不相关的配置；
  • 所有子任务都已经有明确的错误转换策略；
  • 调用方只需要结果列表，不需要管理复杂的生命周期；
  • 任务数量小且有明确上限。

真正应该避免的是：把 gather() 当成任务管理器，然后在它外面不断补取消、超时、重试、队列和清理逻辑。到了那个程度，代码其实已经在手写一个不完整的 TaskGroup 了。

最后的判断标准

一段异步代码是否成熟，不是看它用了多少个 create_task()，也不是看基准测试快了多少。可以用下面的问题检查它：

  1. 一个任务失败时，兄弟任务的命运是否明确？
  2. 上游速度超过下游时，内存是否有上限？
  3. 超时发生后，底层任务是否真的停止并完成清理？
  4. Ctrl+C 或外部取消能否到达最内层？
  5. 是否区分了业务失败、可重试失败和系统失败？
  6. queue.join() 是否一定能结束？
  7. 任务、连接池、文件和临时资源是否拥有清楚的所有者？
  8. 关闭流程是否可以重复执行而不产生第二次破坏？

如果这些问题没有答案，程序可能只是“现在能跑”；如果答案写在代码结构里，程序才开始变得可维护。

异步编程最让人放心的时刻，不是任务同时完成，而是按下停止以后，它知道自己应该怎样停下来。

参考资料

把 asyncio 写到能停下来：TaskGroup、取消、超时与背压
2026/08/05/python-asyncio-control/
Author
Chen Xi
Published
2026-08-05 22:20
License
  1. 版本范围
  2. 一个能跑，但不够可靠的版本
  3. 先理解取消：它是请求，不是按钮
  4. TaskGroup：把任务放进同一个生命周期
  5. 超时应该包住哪一段
  6. 有界队列：背压是稳定性功能
  7. 一个完整的可控任务处理器
    1. 业务错误不一定要取消整组任务
    2. task_done() 必须与 get() 配对
    3. 正常关闭和立即停止是两种不同语义
  8. 外部取消如何传到最里面
  9. 重试不应该和取消混在一起
  10. 用指标观察“可控”是否真的成立
  11. gather() 什么时候仍然合适
  12. 最后的判断标准
  13. 参考资料
  1. 版本范围
  2. 一个能跑，但不够可靠的版本
  3. 先理解取消：它是请求，不是按钮
  4. TaskGroup：把任务放进同一个生命周期
  5. 超时应该包住哪一段
  6. 有界队列：背压是稳定性功能
  7. 一个完整的可控任务处理器
    1. 业务错误不一定要取消整组任务
    2. task_done() 必须与 get() 配对
    3. 正常关闭和立即停止是两种不同语义
  8. 外部取消如何传到最里面
  9. 重试不应该和取消混在一起
  10. 用指标观察“可控”是否真的成立
  11. gather() 什么时候仍然合适
  12. 最后的判断标准
  13. 参考资料