动态创建永久运行的asyncio任务：最佳“即发即弃”方案问询

嘿，针对你这种需要创建大量永久运行asyncio任务的定时程序场景，我来分享几个靠谱的方案，完美避开未await任务被垃圾回收的坑：

方案一：用全局集合跟踪任务（最通用） #

因为asyncio的Task如果没有被任何变量引用，就会被GC回收，所以我们可以用一个全局集合来“hold住”所有永久任务，确保它们不会被回收。同时还能顺便做任务的生命周期管理：

import asyncio
from typing import Set

# 全局集合，用来存所有正在运行的永久任务
running_permanent_tasks: Set[asyncio.Task] = set()

async def permanent_task(task_id: int):
    try:
        while True:
            # 这里写你的任务逻辑，比如定时执行的操作
            print(f"定时任务 {task_id} 执行中...")
            await asyncio.sleep(3600)  # 模拟1小时执行一次
    except asyncio.CancelledError:
        # 处理任务被取消的情况
        print(f"任务 {task_id} 被取消")
    finally:
        print(f"任务 {task_id} 结束")

def spawn_permanent_task(task_id: int):
    task = asyncio.create_task(permanent_task(task_id))
    # 把任务加入集合，防止被GC回收
    running_permanent_tasks.add(task)
    # 任务结束（不管正常还是异常）时，自动从集合移除，避免内存泄漏
    task.add_done_callback(running_permanent_tasks.discard)

这个方案的好处是：

• 简单易实现，兼容所有Python 3.7+版本
• 可以随时通过集合查看当前运行的任务数量和状态
• 自动清理结束的任务，不会造成内存泄漏

方案二：让主循环await所有任务的聚合 #

如果你的主程序就是专门用来运行这些永久任务的，可以把所有任务用asyncio.gather打包，让主循环一直await这个聚合对象，这样所有任务都会被持续跟踪：

import asyncio

async def permanent_task(task_id: int):
    while True:
        print(f"任务 {task_id} 运行中...")
        await asyncio.sleep(3600)

async def main():
    # 批量创建10个永久任务
    task_list = [asyncio.create_task(permanent_task(i)) for i in range(10)]
    # await所有任务（因为是永久任务，这个await会一直挂起，主循环不会退出）
    await asyncio.gather(*task_list)

if __name__ == "__main__":
    asyncio.run(main())

这个方案适合任务数量固定、主程序不需要做其他逻辑的场景，不用额外维护集合，gather会自动帮你跟踪所有任务。

方案三：使用TaskGroup（Python 3.11+ 推荐） #

Python 3.11引入了asyncio.TaskGroup，这是官方推荐的现代任务管理方式，它会自动跟踪所有创建的任务，处理任务的生命周期，甚至能统一处理错误：

import asyncio

async def permanent_task(task_id: int):
    try:
        while True:
            print(f"任务 {task_id} 执行中...")
            await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print(f"任务 {task_id} 被取消")

async def main():
    async with asyncio.TaskGroup() as tg:
        # 批量创建永久任务，TaskGroup会自动跟踪
        for i in range(10):
            tg.create_task(permanent_task(i))
        # TaskGroup会自动等待所有任务完成，这里因为是永久任务，会一直运行

if __name__ == "__main__":
    asyncio.run(main())

注意：默认情况下，如果TaskGroup里的一个任务抛出未捕获的异常，它会取消所有其他任务并退出。如果你想让单个任务崩溃不影响全局，可以在任务内部把所有异常都捕获处理掉。

一些额外注意事项 #

• 错误处理：永久任务一定要加try-except捕获异常，不然一个任务崩溃可能导致整个程序挂掉（尤其是用TaskGroup或者gather的场景）
• 内存监控：如果任务数量非常多，定期检查集合或者TaskGroup的任务状态，确保没有大量僵尸任务占用内存
• 任务取消：如果需要停止任务，不要直接强制杀死，最好用task.cancel()并在任务内部处理CancelledError，优雅退出

内容的提问来源于stack exchange，提问作者OmarZ