读书笔记 · 约 15,168 字 · 预计推送 2 部分

深度精读：Python asyncio 实战——取消、超时、背压、资源清理和常见死锁（主题研究） 未开始

目录

题记与说明

一、来源清单与阅读状态

#来源阅读深度状态
1Coroutines and Tasks — Python 官方文档全文精读✅ 含 Task cancellation / TaskGroup / gather / shield / timeout / wait_for / Task.uncancel() 全节
2Queues — Python 官方文档全文精读✅ 含 maxsize / join / task_done / shutdown(3.13)
3Streams — Python 官方文档全文精读✅ 含 drain() 流控、close()/wait_closed()
4Asyncio Deadlocks in Python — SuperFastPython全文精读✅ 四类死锁完整代码示例 + 三条防御建议
5Queues, Producers and Backpressure — VizLearn全文精读✅ 背压与优雅关闭的完整叙述
6CPython GH-86296: wait_for can hide cancellation in a rare race（原 BPO 42130）标题+迁移后元数据⚠️ 完整评论与修复 PR 正文未读
7Asyncio Coroutine Patterns: Errors and Cancellation — Medium仅搜索摘要❌ 抓取失败（网络错误），结论未经原文核实
8SuperFastPython 背压文（asyncio-queue-backpressure/）—❌ 站点已关：主页公告 "The website SuperFastPython.com has shut down… RIP SuperFastPython November 2021 to March 2026"（2026-03 停运，教程存档另行提供）

检索过程：4 轮 WebSearch（取消/uncancel、timeout vs wait_for、背压/drain/死锁、结构化并发），共覆盖官方文档、CPython issue、SuperFastPython、VizLearn、Medium、Stack Overflow、discuss.python.org 等候选源，最终精读上表来源。

二、七维提取

1. 核心问题

asyncio 是单线程协作式调度：没有操作系统替你抢占、回收、限时，于是传统多线程里"免费"得到的东西全部变成代码结构责任——

2. 主要主张

  1. 取消是协作的、可清理的：官方口径是 try/finally 做清理；若显式捕获 CancelledError，清理完必须重新抛出。"The asyncio components that enable structured concurrency, like asyncio.TaskGroup and asyncio.timeout(), are implemented using cancellation internally and might misbehave if a coroutine swallows asyncio.CancelledError." 确要吞取消，必须同时调 Task.uncancel() 复位取消计数（3.11+）。
  2. 超时应统一用 asyncio.timeout()（3.11+）而非 wait_for：wait_for 超时会取消内层任务但总等待时长可能超过 timeout（因为"will wait until the future is actually cancelled"），且有已知竞态 GH-86296（外部取消可被误当超时吞掉）；3.12 起 wait_for 干脆改用 asyncio.timeout() 实现。
  3. 背压 = 有界缓冲：Queue(maxsize=N) 的 await put() 满即挂起是生产端节流；StreamWriter.drain() 的高/低水位是写出端节流。无界队列（maxsize=0）把背压问题延迟成内存问题。
  4. 清理可靠性来自结构而非小心：async with + try/finally + TaskGroup（一个子任务失败即取消其余并聚合成 ExceptionGroup）；队列消费端 task_done() 必须放 finally，否则 join() 永挂；3.13 的 Queue.shutdown() 解决"关闭时生产者仍阻塞在 put"的关闭死锁。
  5. 死锁模式有限且可枚举：等自己（非重入锁二次获取，含跨函数嵌套）、互等环、锁序不一致、异常路径不释放锁、join() 缺 task_done()、缺关闭信号；防御三件套 = 上下文管理器 + 等待可超时 + 全局锁序。

3. 方法（本主题研究的论证路径）

4. 数据与证据

关键原文引述（全部来自本 session 抓取的全文）：

取消与清理（官方 tasks 文档）：

"Tasks can easily and safely be cancelled. When a task is cancelled, asyncio.CancelledError will be raised in the task at the next opportunity. It is recommended that coroutines use try/finally blocks to robustly perform clean-up logic."
"If end-user code is, for some reason, suppressing cancellation by catching CancelledError, it needs to call this method (uncancel()) to remove the cancellation state."

gather 的著名陷阱（同上）：

"If return_exceptions is False (default), the first raised exception is immediately propagated to the task that awaits on gather(). Other awaitables in the *aws* sequence won't be cancelled and will continue to run."

TaskGroup 的结构化保证（同上）：

"The first time any of the tasks belonging to the group fails with an exception other than asyncio.CancelledError, the remaining tasks in the group are cancelled."

任务弱引用（同上，create_task 的 Important 框）：

"The event loop only keeps weak references to tasks. A task that isn't referenced elsewhere may get garbage collected at any time, even before it's done."

wait_for 的超时语义（同上）：

"If a timeout occurs, it cancels *aw* and raises TimeoutError. … The function will wait until the future is actually cancelled, so the total wait time may exceed the *timeout*."（3.7 起行为；"Changed in version 3.12: Implemented using asyncio.timeout()"）

背压（官方 queues 文档）：

"If *maxsize* is less than or equal to zero, the queue size is infinite. If it is an integer greater than 0, then await put() blocks when the queue reaches *maxsize* until an item is removed by get()."
"Note that methods of asyncio queues don't have a *timeout* parameter; use asyncio.wait_for() function to do queue operations with a timeout."

Queue.shutdown（3.13）（同上）：

"Future calls to put() raise QueueShutDown. Currently blocked callers of put() will be unblocked and will raise QueueShutDown in the formerly awaiting task."

写出端流控（官方 streams 文档，drain()）：

"This is a flow control method that interacts with the underlying IO write buffer. When the size of the buffer reaches the high watermark, *drain()* blocks until the size of the buffer is drained down to the low watermark and writing can be resumed."
"When the write buffer is below the high watermark, drain() returns immediately without yielding to the event loop. As a result, code which repeatedly calls write() followed by await drain() may prevent other tasks from running."

连接关闭（同上，wait_closed()）：

"Should be called after close() to wait until the underlying connection is closed, ensuring that all data has been flushed before e.g. exiting the program."

死锁定义与成因（SuperFastPython，引 The Art of Concurrency, 2009 术语表）：

"Deadlock: A situation in which one or more threads are waiting for an event that will never occur."

成因清单（原文）：waits on itself（同一锁获取两次）、wait on each other（await 环）、fails to release（异常路径不释放锁/信号量）、locks acquired in different orders。防御（原文要点）："Use context managers when acquiring and releasing locks. / Use timeouts when waiting. / Always acquire locks in the same order."，并指出 asyncio 同步原语"methods of these synchronization primitives do not accept the timeout argument"（引官方 sync 原语文档），所以等待超时必须外包 wait_for。

背压的定性（VizLearn）：

"maxsize is the backpressure. A full queue suspends put, which paces the producer to the consumer's rate automatically."
"Unbounded means the producer never waits, and a fast producer then grows the queue until memory runs out."
"task_done is in a finally. If the work raises and the call is skipped, join waits forever for an item that has been taken but never acknowledged, and the program hangs at the end with no error."
"Why does my program hang at the end? Almost always a missing task_done on some path, so join is still waiting; or consumers blocked on await q.get() with nothing left to arrive and no cancellation. Both look like a deadlock and are really a missing shutdown signal."

真实缺陷案例（GH-86296，BPO 42130）：
标题即结论："AsyncIO's wait_for can hide cancellation in a rare race condition"；2020-10-23 由 tvoinarovskyi 报告，波及 3.8/3.9/3.10，labels type-bug/topic-asyncio，PR #26097、#28149，关联修复 PR GH-98607 "Fix for asyncio.wait_for() swallowing cancellation, and add tests"。与官方文档"3.12 起用 asyncio.timeout() 重实现 wait_for"互为印证。

本机实验（见第五节）：5/5 符合文档声明。

5. 结论

asyncio 生产可靠性可以压缩成一张结构性规则表，满足后五类故障在结构上被排除，而不是靠运行时运气：

  1. 生命周期：优先 TaskGroup（失败传播 + 取消其余 + 强引用）；裸 create_task 必须自存强引用并处理异常（否则弱引用可被 GC、"Task exception was never retrieved"）。
  2. 取消：try/finally 清理；捕获 CancelledError 后重抛；吞取消必须配 uncancel()；shield() 只用于"必须完成的清理/提交"，且注意外层仍会收到 CancelledError。
  3. 超时：新代码一律 asyncio.timeout() / timeout_at()（支持 reschedule、可安全嵌套）；wait_for 注意"总时长可超限"与历史吞取消竞态；asyncio.wait()/as_completed() 超时不取消剩余任务（要么有意用之，要么记得善后）。
  4. 背压：每条生产-消费链路显式 maxsize；task_done() 放 finally；宁可 put_nowait+计数主动丢负载也不要无界；写出端用 write()+await drain() 并警惕"水位以下 drain 不让出循环"。
  5. 死锁防御：非重入锁绝不二次获取（含跨函数）；锁全局定序；一切等待可超时；优雅关闭用 Queue.shutdown()/显式取消，而不是依赖"恰好没人再 put"。

6. 局限

7. 对我的启发（可执行检查单）

写 Agent 工具调用、爬虫、日志/数据管道这类 async 代码时：

  1. 每个后台任务三问：谁持强引用？异常谁取？取消时 finally 干什么？——三者缺一就是"看起来能跑"的地雷。
  2. LLM/HTTP 等慢 IO 一律 async with asyncio.timeout(...)：别用 wait_for（历史坑 + 3.12 前吞取消竞态），也别指望库自带的 connect timeout 覆盖整个请求周期。
  3. 多级管道每级队列给 maxsize：背压是防 OOM 的唯一机制；except QueueFull: dropped += 1 是与背压相反的合法策略（丢负载），必须显式选择并计数。
  4. except Exception 不会误吞取消（3.8+ 起 CancelledError 继承 BaseException）——但要防 except BaseException 和裸 except:；写了就检查是否重抛。
  5. "程序退出时挂住"排查顺序：缺 task_done 的路径 → 无退出条件的 await q.get() 循环 → 关闭时阻塞在 put()/drain() 无人放行（用 Queue.shutdown()）→ 后台任务弱引用被 GC 的假象。
  6. 三条死锁军规进 code review checklist：锁上下文管理器化、所有等待可超时、锁全局定序。
  7. gather(..., return_exceptions=False) 默认会把兄弟任务留在野外——要么 return_exceptions=True，要么 TaskGroup；二者取舍是语义决策不是随手写的。

三、分主题详注

3.1 取消：从注入点到 uncancel()

3.2 超时：timeout() vs wait_for vs wait()

3.3 背压：两条独立的节流链

3.4 资源清理

3.5 常见死锁（四模式 + 三军规）

  1. 等自己：非重入 asyncio.Lock 二次获取（直接嵌套，或 task1 持锁后 await task2(lock)）。实验 C 复现：0.3s 内永不完成。
  2. 互等环：A await B、B await A（SuperFastPython 用 main 与新 task 互等演示）；任何等待原语（锁/信号量/条件/队列间接）都可能成环。
  3. 锁序：两个协程以不同顺序获取两把锁 → 循环等待；解法锁序（lock ordering）。
  4. 不释放：临界区内抛异常且没用上下文管理器 → 锁永久被占（示例输出里连 "Task exception was never retrieved" 都先出现了）。
  5. 队列尾部信号缺失（VizLearn）：join() 等一个被取走却没人 task_done 的项 → 程序结尾无错挂死；或消费者停在 await q.get() 无退出条件。
  6. 三军规（SuperFastPython）：上下文管理器持锁；一切等待可超时（原语自身无 timeout 参数，外包 wait_for）；锁全局定序。

四、两个死锁相关语义的交叉印证

五、本机验证实验（Python 3.12.12，`python -I asyncio_check.py`）

[A] cleanup ran on cancellation: True; task.cancelled()=True
    → 取消时 finally 清理先于 CancelledError 逸出（官方 try/finally 建议的前提成立）
[B] caught CancelledError, cancelling()=1
[B] after uncancel(), cancelling()=0, result=survived
    → 取消计数机制与文档一致：吞取消 + uncancel() 可复位到 0（演示用，生产不鼓励）
[C] lock re-acquisition hung past 0.3s -> TimeoutError (deadlock confirmed), t.done()=False
    → 非重入锁二次获取即死锁；wait_for 超时可把"挂死"变成可处理的 TimeoutError
[D] put 0 (qsize=1) / put 1 (qsize=2) / got 0 / put 2 (qsize=2) / got 1 / ...
    → maxsize=2 下 put 2 一直挂到消费者取走 0：背压确实发生在 put 阻塞上
[E] timeout raised TimeoutError after 0.20s; outer code still running
    → asyncio.timeout(0.2) 到期转 TimeoutError，且不影响外层代码（结构化边界成立）

六、失败与不可达记录（按规范单独列出，不计入精读成果）

  1. Medium《Asyncio Coroutine Patterns: Errors and Cancellation》：web_reader 返回 500 网络错误（error id: 202611010846077f36b15f58594fd7_call_b9f8b203f3f2455a84cf0353），未获原文；其结论仅以搜索摘要形式出现于第二节之外的辅助印象，未作为本笔记任何独立主张的依据。
  2. SuperFastPython《Backpressure with Bounded Queues》（/asyncio-queue-backpressure/）：站点 2026-03 关停，返回停运公告页；背压部分改由官方 queues/streams 文档与 VizLearn 承担。
  3. GH-86296 完整评论流、bpo-42130 原始页面、修复 PR #98607 正文：未读取。
  4. "put 满 + drain 堵"环形死锁：未写出复现代码（时间与调度窗口成本），保持推断标注。

七、来源链接汇总

← 资料库今日播报 →