异步main函数启动AsyncIOScheduler定时任务报AssertionError如何解决

报错原因 #

你遇到的AssertionError是因为asyncio.run()会自动创建、管理事件循环，执行传入的协程时循环已经处于运行状态。你在main协程内部调用asyncio.get_event_loop().run_forever()，相当于在已运行的循环中重复启动，触发了框架内部的断言校验。

两种可行的修改方案都不需要重构整体业务结构，既可以先异步执行TestApiConnection拿到结果，也能正常启动定时任务：

方案1：用阻塞等待替代手动启动循环 #

AsyncIOScheduler启动后会自动绑定当前运行的事件循环，不需要手动调用run_forever()，只要增加永久等待的逻辑让main协程不退出即可：

import asyncio
import datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def TestApiConnection():
    # 保留原有测试逻辑
    pass

async def QueryData(): 
    print(datetime.datetime.now())

async def main():
    await TestApiConnection()
    
    scheduler = AsyncIOScheduler(timezone="Europe/Berlin")
    scheduler.add_job(QueryData, 'cron', minute='0-59')
    scheduler.start()
    
    # 永久阻塞保持协程运行
    await asyncio.Event().wait()
    
    try:
        while True:
            await asyncio.sleep(3600)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown(wait=False)


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

方案2：手动管理事件循环，不使用asyncio.run #

如果需要更灵活的循环控制，可以放弃asyncio.run()，自行创建、启动事件循环避免冲突：

import asyncio
import datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def TestApiConnection():
    # 保留原有测试逻辑
    pass

async def QueryData(): 
    print(datetime.datetime.now())

async def main():
    await TestApiConnection()
    
    scheduler = AsyncIOScheduler(timezone="Europe/Berlin")
    scheduler.add_job(QueryData, 'cron', minute='0-59')
    scheduler.start()


if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    scheduler = None
    try:
        loop.run_until_complete(main())
        loop.run_forever()
    except (KeyboardInterrupt, SystemExit):
        pass
    finally:
        # 关闭调度器和循环
        for task in asyncio.all_tasks(loop):
            coro = task.get_coro()
            if hasattr(coro, "__self__") and isinstance(coro.__self__, AsyncIOScheduler):
                coro.__self__.shutdown(wait=False)
                break
        loop.close()

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