如何用Python协程泄漏的自动检测工具定位异步资源问题

Python异步程序在长周期运行中，如果协程未被正确关闭或任务未被await，就会产生协程泄漏，表现为内存持续增长、事件循环负载变高。借助协程泄漏的自动检测工具，可以在不修改业务代码太多的情况下发现这些隐患。

协程泄漏的常见原因

协程泄漏通常不是语法错误，而是逻辑上的资源未释放。下面列出几种典型情况：

• 创建了任务但没有挂载到事件循环外的作用域，导致引用丢失却未取消
• 在异常处理中漏掉await，协程对象被垃圾回收但并未执行完
• 使用asyncio.Queue等结构时，生产者协程退出而消费者未结束

自动检测工具的工作方式

大多数检测工具通过替换事件循环中的协程创建函数，或者利用gc模块与weakref来追踪存活协程。当程序退出或定时检查时，若仍存在预期外的协程对象，就输出警告。

基于weakref的简单监控示例

我们可以用弱引用集合记录所有协程，再定期打印未结束的协程数量。

import asyncio
import weakref

# 用弱引用集合保存协程对象
_active = weakref.WeakSet()

def track(coro):
    # 将协程加入监控集合
    _active.add(coro)
    return coro

async def sample_task():
    await asyncio.sleep(1)

async def main():
    # 手动通过track包装协程
    t = track(asyncio.create_task(sample_task()))
    await asyncio.sleep(0.1)
    # 此时sample_task尚未完成，应检测到1个活跃协程
    print("active coroutines:", len(_active))

asyncio.run(main())

利用loop.set_task_factory钩子

更自动化的方式是替换任务工厂，让所有任务自动被记录。

import asyncio
import weakref

_tasks = weakref.WeakSet()

def make_factory(loop):
    # 保存原始任务工厂
    base = loop.get_task_factory()
    def factory(loop, coro, *a, **k):
        # 创建任务并加入弱引用集合
        task = base(loop, coro, *a, **k) if base else asyncio.Task(coro, loop=loop)
        _tasks.add(task)
        return task
    return factory

async def job():
    await asyncio.sleep(2)

async def main():
    loop = asyncio.get_event_loop()
    loop.set_task_factory(make_factory(loop))
    asyncio.create_task(job())
    await asyncio.sleep(0.5)
    print("leaked tasks:", len(_tasks))

asyncio.run(main())

在测试环境集成报警

在pytest或unittest中，可在用例结束后检查监控集合。若还有未关闭协程，则判定为泄漏并失败。

使用现有工具的建议

如果不想自己实现，可关注一些开源方案，它们通常提供装饰器或pytest插件。关键在于把检测放在持续集成中，而不是上线后靠用户反馈。通过协程泄漏的自动检测工具，团队能在早期暴露异步资源管理问题，减少线上故障。

注意：在生产环境长期开启详细追踪可能影响性能，建议仅在预发或测试环境默认启用。

Python协程泄漏自动检测工具修改时间：2026-07-25 18:39:15