协程泄漏主因是Task未取消或等待完成且被引用，导致GC不回收；应监控all_tasks()数量变化、用objgraph查增长、测试中用fixture清理、生产环境结合滑动窗口与白名单告警。
协程对象没被 gc 掉，asyncio.Task 仍在运行怎么办
Python 协程泄漏最典型的表征就是 asyncio.Task 数量持续上涨，但对应逻辑早已该结束。根本原因不是“写了 async 就会泄漏”，而是任务没被显式取消或等待完成，又没被引用释放——尤其在异常提前退出、超时丢弃、或忘记 await 的地方。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 用
asyncio.all_tasks()在关键检查点（如测试 teardown、服务健康接口）抓快照，对比前后数量变化 - 对每个
Task调用task.get_coro()和task.get_stack()查看挂起点，定位未完成的协程源头 - 避免直接用
asyncio.create_task()后不管：要么加try/finally确保await task或task.cancel()，要么用asyncio.timeout()包裹 - 注意：
asyncio.create_task()返回的对象若被变量引用（哪怕只是临时赋值），GC 不会回收它，即使协程已结束
tracemalloc 能不能查协程泄漏
不能直接查。tracemalloc 跟踪的是内存分配堆栈，而协程对象本身很小，真正占资源的是它持有的上下文、闭包变量、或阻塞中的 IO 对象（比如没关闭的 aiohttp.ClientSession）。靠它只能间接发现“某段异步代码反复分配大对象”，但无法确认是协程没结束还是数据堆积。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 开启
tracemalloc前先调用asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())（Windows）或确保 loop 已启动，否则可能报RuntimeError: no running event loop - 配合
sys.getrefcount()检查特定Task引用数是否异常高（比如 >2 且长期不降） - 更有效的是用
objgraph：安装后执行objgraph.show_growth(limit=10)，重点观察Task、coroutine、Future类型的增长趋势
用 pytest-asyncio 写测试时怎么防泄漏
测试框架默认不清理未完成的 Task，一个 test 函数里漏掉 await 或抛出异常中断，就可能让 Task 泄漏到下一个 test 里，造成干扰甚至失败。
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成，提供 184 种工具和 200 多条命令，用于安全分析和代码质量...
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 在
conftest.py中加 fixture，用yield前后比对asyncio.all_tasks()： - 确保
pytest-asyncio配置了asyncio_mode = "auto"，并禁用--asyncio-mode=strict（它只校验签名，不保证执行） - 测试中避免裸写
asyncio.create_task(some_coro())；改用asyncio.create_task(some_coro(), name="test_xxx")，便于后续按名过滤排查 - 如果用了
asyncio.sleep(0)触发调度，记得之后await asyncio.wait_for(task, timeout=0.1)，别让它悬着
生产环境用 psutil + asyncio 监控 Task 数量够不够
够用，但有盲区。单纯监控 len(asyncio.all_tasks()) 只能告诉你“现在有多少 Task”，无法区分是正常并发还是泄漏。如果服务长期稳定在 50±5，突然涨到 200 并持续，才值得告警。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 每 10 秒采集一次
len(asyncio.all_tasks())，同时记录len([t for t in asyncio.all_tasks() if not t.done()])—— 后者才是真正在跑的 - 结合
psutil.Process().memory_info().rss看内存是否同步上涨，排除是 Task 持有大量数据导致的假阳性 - 不要只依赖单点数值：用滑动窗口计算标准差，当连续 3 次超出均值 + 2σ 时触发检查，比固定阈值靠谱
- 注意：某些后台 Task（如心跳、日志刷盘）本就不
done，需白名单过滤，否则天天告警
真正难的不是发现泄漏，是判断哪个 Task 该结束却没结束——它可能卡在第三方库的 await 上，也可能被一个没暴露 cancel 接口的底层 Future 拖住。这时候得进源码翻 __await__ 实现，或者用 asyncio.debug = True 开启事件循环调试模式，看它卡在哪一行。
