如何在asyncio.gather执行后动态添加异步任务并支持永久轮询？

问题分析 #

你现在的代码有几个明显问题：

1. 类方法定义漏了self参数，根本没法正常访问实例里的属性；
2. asyncio.gather(*tasks)只会等调用它时传进去的那些任务，后面往self.tasks加的新任务，这个gather根本不会管——要是原任务都是永久循环的，main会一直卡着，但不会主动等新任务；
3. 你外部调用ExampleClass.add_new_task(task)的方式错了，这是实例方法不是类方法，得先创建类的实例才能调用。

针对你要动态加永久任务的需求，给你两种靠谱的实现方式：

方案1：用Python 3.11+的asyncio.TaskGroup（推荐） #

Python 3.11出的TaskGroup专门解决这种动态任务的场景，它能自动管理所有加进去的任务，还会等所有任务完成，用起来特别省心。

import asyncio

async def f1(num):
    while True:
        print(num)
        await asyncio.sleep(2)

class ExampleClass:
    def __init__(self):
        self.task_group = None

    async def main(self):
        # 启动TaskGroup上下文
        async with asyncio.TaskGroup() as tg:
            self.task_group = tg
            # 先创建10个永久任务
            for i in range(10):
                tg.create_task(f1(i))
            # 加个永久阻塞的Future，让main一直跑着，保持TaskGroup存活
            await asyncio.Future()

    def add_new_task(self, coro):
        # 传协程对象进来，TaskGroup会自动把它变成任务并调度执行
        if self.task_group:
            self.task_group.create_task(coro)

async def run():
    example = ExampleClass()
    await example.main()

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

外部加任务的正确方式（得确保已经创建了ExampleClass的实例，而且main已经在事件循环里跑起来了）：

# 假设已经有了example实例
example.add_new_task(f1(10))

方案2：兼容Python 3.10及以下版本（手动管任务集合） #

要是你用的Python版本低于3.11，那就手动维护一个任务集合，定期清理完成的任务，同时用asyncio.wait等着所有任务。

import asyncio
from typing import Set

async def f1(num):
    while True:
        print(num)
        await asyncio.sleep(2)

class ExampleClass:
    def __init__(self):
        self.tasks: Set[asyncio.Task] = set()

    async def main(self):
        # 初始化10个永久任务
        for i in range(10):
            task = asyncio.create_task(f1(i))
            self.tasks.add(task)
            # 加个回调，任务完成后自动从集合里删掉（永久任务不会触发这个）
            task.add_done_callback(self.tasks.discard)
        
        # 一直等着所有任务，要是有任务意外完成就清理掉
        while self.tasks:
            done, pending = await asyncio.wait(
                self.tasks,
                return_when=asyncio.FIRST_COMPLETED
            )
            for task in done:
                self.tasks.discard(task)

    def add_new_task(self, coro):
        # 把协程转成任务，加到集合里
        task = asyncio.create_task(coro)
        self.tasks.add(task)
        task.add_done_callback(self.tasks.discard)

async def run():
    example = ExampleClass()
    await example.main()

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

外部加任务的方式：

example.add_new_task(f1(10))

注意点 #

• 永久任务会一直占着事件循环资源，要是以后需要停任务，最好给任务加个退出开关（比如用asyncio.Event控制循环停止）；
• 加任务的时候，一定要传协程对象，然后用create_task或者TaskGroup.create_task把它加到事件循环里，不然协程根本不会跑；
• 两种方式都能满足你的需求，TaskGroup更简洁，是官方推的新特性，能用的话优先选这个。

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