如何在Python生产环境下调试异步死锁与任务阻塞问题？

生产环境禁用 asyncio.run()，须手动管理事件循环并捕获未完成任务；asyncio.Lock 必须加超时；create_task() 后需显式跟踪等待；连接池耗尽等资源问题亦会导致无日志卡死。

asyncio.run() 在生产环境直接调用会掩盖死锁信号

生产环境不能用 asyncio.run() 启动主事件循环——它会在异常未处理时静默退出，或在任务未完成时强制 cancel 并吞掉 RuntimeError: Event loop is closed 这类关键线索。真正死锁发生时，进程可能卡住不动，但日志里什么都没有。

正确做法是手动管理事件循环，并捕获未完成任务：

import asyncio
import signal
import sys
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
注册 SIGUSR1（Linux/macOS）或 SIGBREAK（Windows）用于诊断
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().qualname):
print(f"  - {t.get_coro().qualname}, state={t._state}")
if sys.platform != "win32":
loop.add_signal_handler(signal.SIGUSR1, dump_pending_tasks)
else:
signal.signal(signal.SIGBREAK, lambda s, f: dump_pending_tasks())
try:
loop.run_until_complete(main())
except KeyboardInterrupt:
pass
finally:
显式检查残留任务，避免 silent hang
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")
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 会被取消，但你根本不知道它们曾卡在哪。
必须显式跟踪和等待关键后台任务：
                                                                        Python 3.14.3
微软官方的 Python 扩展，是 VS Code 安装量最高的扩展（209M+）。集成 IntelliSense（通过 Pylance）、调试（通过 Python Debugger）、代码检查、格式化、重构和单元测试等功能。支持 Jupyter Notebook、虚拟环境管理和多 Python 版本切换。

用 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、连接池、锁、信号处理几层叠加后，某条路径上既没超时、也没异常、还没日志——最危险的卡死，恰恰发生在你认为“它应该能自己恢复”的地方。