Asyncio多任务调度异常：后台任务未正确加入事件循环

1. 异步函数未被正确调用
   schedule_task是用async def定义的异步函数，但在select_task中调用时未添加await关键字，导致协程对象从未被调度执行，触发RuntimeWarning: coroutine 'aRobot.schedule_task' was never awaited警告。

2. 事件循环不统一
   初始化时手动创建了self.loop = asyncio.new_event_loop()，但程序实际运行依赖asyncio.run()创建的默认事件循环。后台任务被提交到未启动的手动循环中，既不会执行，也无法被asyncio.all_tasks()（默认获取当前运行循环的任务）检测到。

3. 任务管理机制缺失
   self.tasks字典未被实际使用，创建的后台任务没有被存储，无法追踪和查看任务状态。

针对上述问题，调整代码如下：

• 删除手动创建的事件循环，使用asyncio.get_running_loop()获取当前运行的循环
• 调用异步函数时添加await关键字
• 利用self.tasks存储后台任务，便于后续查看和管理
• 优化start方法的逻辑，避免不必要的TaskGroup嵌套

修正后的代码：

import asyncio
from uuid import uuid4

class aRobot:
    def __init__(self) -> None:
        self.isrunning = True
        self.tasks = {}  # 存储任务：key为任务ID，value为task对象

    async def asleep(self) -> None:
        await asyncio.sleep(20)
        print("asleep completed")

    async def start(self) -> None:
        while self.isrunning:
            await self.menu()
            await self.get_input()

    async def menu(self) -> None:
        print("\n1. Add task1")
        print("3. Check all tasks")
        print("5. Exit")

    async def get_input(self):
        try:
            usr_input = input('==> ')
            await self.select_task(int(usr_input))
        except ValueError:
            print("无效输入，请输入数字选项")

    async def select_task(self, input: int):
        if input == 1:
            await self.schedule_task(self.asleep())
        elif input == 3:
            await self.check_tasks()
        elif input == 5:
            self.isrunning = False
            print("退出程序")
        else:
            print("无效选项，请重新输入")

    async def schedule_task(self, coro):
        loop = asyncio.get_running_loop()
        task = loop.create_task(coro)
        task_id = str(uuid4())[:8]  # 生成短任务ID
        self.tasks[task_id] = task
        # 任务完成后从字典中移除
        def task_done_callback(fut):
            self.tasks.pop(task_id, None)
        task.add_done_callback(task_done_callback)
        print(f"已添加任务：{task_id}")

    async def check_tasks(self):
        if not self.tasks:
            print("当前无后台任务")
            return
        print("当前后台任务：")
        for task_id, task in self.tasks.items():
            status = "运行中" if not task.done() else "已完成"
            print(f'> 任务ID: {task_id}, 状态: {status}, 协程: {task.get_coro()}')

if __name__ == "__main__":
    robot = aRobot()
    asyncio.run(robot.start())

• 事件循环统一：使用asyncio.get_running_loop()获取当前由asyncio.run()管理的事件循环，确保后台任务在正确的循环中执行。
• 异步函数调用：在select_task中调用await self.schedule_task(...)，确保协程被正确调度。
• 任务管理：通过self.tasks字典存储任务，并添加完成回调自动清理已完成的任务，方便查看任务状态。
• 输入容错：增加ValueError捕获，处理非数字输入的情况，提升程序健壮性。

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