生产环境禁用 asyncio.run()，须手动管理事件循环并捕获未完成任务；asyncio.lock 必须加超时；create_task() 后需显式跟踪等待；连接池耗尽等资源问题亦会导致无日志卡死。
asyncio.run() 在生产环境直接调用会掩盖死锁信号
生产环境不能用 asyncio.run() 启动主事件循环——它会在异常未处理时静默退出，或在任务未完成时强制 cancel 并吞掉 RuntimeError: Event loop is closed 这类关键线索。真正死锁发生时，进程可能卡住不动，但日志里什么都没有。
正确做法是手动管理事件循环，并捕获未完成任务：
import asyncio
import signal
import sys
<p>loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)</p><h1>注册 SIGUSR1（Linux/macOS）或 SIGBREAK（Windows）用于诊断</h1><p>def dump_pending_tasks():
pending = asyncio.all_tasks(loop)
print(f"[DEBUG] {len(pending)} pending tasks:")
for t in sorted(pending, key=lambda x: x.get_coro().<strong>qualname</strong>):
print(f" - {t.get_coro().<strong>qualname</strong>}, state={t._state}")</p><p>if sys.platform != "win32":
loop.add_signal_handler(signal.SIGUSR1, dump_pending_tasks)
else:
signal.signal(signal.SIGBREAK, lambda s, f: dump_pending_tasks())</p><p>try:
loop.run_until_complete(main())
except KeyboardInterrupt:
pass
finally:</p><h1>显式检查残留任务，避免 silent hang</h1><pre class="brush:python;toolbar:false;">remaining = asyncio.all_tasks(loop)
if remaining:
print(f"[FATAL] {len(remaining)} tasks still running at shutdown")
for t in remaining:
t.cancel()
loop.run_until_complete(asyncio.gather(*remaining, return_exceptions=True))
loop.close()
await asyncio.Lock() 阻塞却无超时，是常见死锁源头
asyncio.Lock 默认不提供超时机制，一旦某个协程持锁后崩溃、未释放，或 await 了另一个也依赖该锁的协程，整个链就卡死。生产环境必须强制加超时，且需区分「获取锁失败」和「锁内逻辑超时」。
- 永远不要写
await lock.acquire()—— 改为await asyncio.wait_for(lock.acquire(), timeout=5.0) - 锁内逻辑也要套
asyncio.wait_for(..., timeout=...)，否则锁占着不放 - 使用
try/finally确保lock.release()执行，但注意：如果acquire()抛出TimeoutError，release()会报RuntimeError: Lock is not acquired
更稳妥的模式：
async def guarded_work(lock, work_coro):
try:
await asyncio.wait_for(lock.acquire(), timeout=3.0)
except asyncio.TimeoutError:
raise RuntimeError("Failed to acquire lock within timeout")
<pre class="brush:python;toolbar:false;">try:
return await asyncio.wait_for(work_coro, timeout=8.0)
finally:
if lock.locked(): # 防止重复 release 或未 acquire 就 release
lock.release()
asyncio.create_task() 后不 await / gather，任务会“消失”在后台
生产代码里常有人写 asyncio.create_task(some_background_job()) 就不管了，以为它会自己跑完。实际上：如果该 task 抛出未捕获异常，asyncio 只会打一行 Task exception was never retrieved 警告然后吞掉；如果主协程结束，这些 task 会被取消，但你根本不知道它们曾卡在哪。
必须显式跟踪和等待关键后台任务：
- 用
asyncio.current_task().get_name()或自定义命名（create_task(..., name="metrics_flush")）便于识别 - 把所有重要 background task 存进集合，shutdown 前
asyncio.gather(*tasks, return_exceptions=True) - 对非关键任务，至少加一层异常兜底：
asyncio.create_task(silent_wrapper(task))，其中silent_wrapper捕获并记录异常，避免污染 event loop
示例兜底封装：
async def silent_wrapper(coro):
try:
return await coro
except Exception as e:
# 记录完整 traceback，包括 task 名
task = asyncio.current_task()
logger.error(f"Background task {task.get_name() or 'unnamed'} failed", exc_info=e)
第三方库（如 httpx、aiomysql）的连接池耗尽也会表现为“await 不返回”
这不是 Python 层面的死锁，而是底层资源枯竭。比如 httpx.AsyncClient 默认连接池上限是 10，若并发发起 20 个请求且响应慢，后 10 个会无限等待空闲连接；aiomysql.Pool 的 maxsize 设太小同理。现象是 await client.get(...) 卡住，但 asyncio.all_tasks() 显示所有 task 都在 wait_for 或 recv 状态。
排查重点：
- 检查客户端初始化参数：
httpx.AsyncClient(limits=httpx.Limits(max_connections=50)) - 确认是否复用 client 实例——每个 request 都 new 一个 client 会快速耗尽文件描述符
- 用
lsof -p PID | grep TCP查看连接数是否接近 ulimit - 对数据库池，启用
echo=True和pool_recycle=3600避免 stale connection
临时诊断可在 await 前加 trace：
import time
start = time.time()
try:
resp = await client.get(url)
logger.debug(f"HTTP GET {url} took {time.time()-start:.2f}s")
except Exception as e:
logger.warning(f"HTTP GET {url} failed after {time.time()-start:.2f}s", exc_info=e)
真实死锁往往不是纯 Python 协程调度问题，而是异步 I/O、连接池、锁、信号处理几层叠加后，某条路径上既没超时、也没异常、还没日志——最危险的卡死，恰恰发生在你认为“它应该能自己恢复”的地方。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
