Python asyncio 长运行服务中任务为何会导致内存持续增长？

2026-11-06数据阿明

结论

示例中的内存持续增长，是因为全局列表 tasks 一直保存着每个 Task 对象。任务完成后，列表仍保留对它的引用，因此任务对象及其结果无法被垃圾回收。示例每 0.5 秒创建一个任务，却从不移除已完成的任务，列表便会不断变大。

内存增长的原因是任务引用没有及时清理。管理任务时，应保留仍在运行的任务，在任务结束后移除引用，并处理任务异常和服务关闭。

管理后台任务

可以用集合追踪尚未完成的任务，在任务结束时将其移除并读取异常。服务关闭时，取消剩余任务并等待它们结束：

import asyncio
import logging

logging.basicConfig(level=logging.INFO)

async def worker(n):
    await asyncio.sleep(1)
    return n * 2

async def scheduler():
    pending = set()

    def on_done(task):
        pending.discard(task)
        if task.cancelled():
            return
        exc = task.exception()
        if exc is not None:
            logging.error(
                "后台任务失败",
                exc_info=(type(exc), exc, exc.__traceback__),
            )

    try:
        while True:
            task = asyncio.create_task(worker(10))
            pending.add(task)
            task.add_done_callback(on_done)
            await asyncio.sleep(0.5)
    finally:
        remaining = tuple(pending)
        for task in remaining:
            task.cancel()
        await asyncio.gather(*remaining, return_exceptions=True)

async def main():
    await scheduler()

asyncio.run(main())

on_done 会在任务结束时将它从集合中移除，并调用 task.exception() 读取异常，避免任务失败被忽略。调度器退出或被取消时，finally 会清理仍在运行的任务。如果希望由调用方统一处理任务失败，也可以等待任务并读取结果，而不是只在回调中记录异常。

让并发量有上限

如果任务的处理时间可能超过调度间隔，持续调用 create_task() 会让同时运行的任务越来越多。即使已完成的任务都能及时移除，工作积压仍可能让服务占用更多内存。对于持续输入的工作负载，通常应设置并发上限，例如用 asyncio.Queue 配合固定数量的 worker。队列也应设置容量，并明确队列满时如何处理。

如果任务是一组有限的子任务，可以考虑用 asyncio.TaskGroup 管理。无限调度循环则需要考虑并发限制和退出时的清理。

检查内存是否真的泄漏

Python 对象不再被引用后可以被垃圾回收，但进程的 RSS 不一定会立刻下降。内存分配器可能会保留已释放的内存，供后续分配使用。可以结合 tracemalloc 比较不同时间点的快照，确认仍被引用的 Python 对象是否持续增加，同时观察未完成任务数和待处理队列长度。强制垃圾回收无法释放仍被 tasks 等容器引用的对象。