AsyncIOScheduler定时任务无法启动问题求助

代码示例：

sheduler = AsyncIOScheduler(timezone='Europe/Moscow')

async def thread_maintaining_communication():
    print('There')

async def main():
    sheduler.add_job(thread_maintaining_communication,"interval", seconds=20)
    sheduler.start()
    #await bot.infinity_polling(skip_pending=True)
    while True:
        print('sleep 10 sec')
        await asyncio.sleep(10)

asyncio.run(main())

控制台输出：

INFO:apscheduler.scheduler:Adding job tentatively -- it will be properly scheduled when the scheduler starts
INFO:apscheduler.scheduler:Added job "thread_maintaining_communication" to job store "default"
sleep 10 secINFO:apscheduler.scheduler:Scheduler started

sleep 10 sec
sleep 10 sec
sleep 10 sec

可见调度器已启动且任务已添加，但定时任务thread_maintaining_communication始终未执行。

1. 显式绑定事件循环 #

AsyncIOScheduler依赖asyncio事件循环运行，显式指定当前事件循环可避免自动获取时的潜在问题：

import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def thread_maintaining_communication():
    print('There')

async def main():
    # 获取当前运行的事件循环
    loop = asyncio.get_running_loop()
    # 绑定事件循环创建调度器
    scheduler = AsyncIOScheduler(timezone='Europe/Moscow', event_loop=loop)
    # 添加任务时指定replace_existing=True，避免重复添加
    scheduler.add_job(thread_maintaining_communication, "interval", seconds=20, replace_existing=True)
    scheduler.start()
    print("调度器已启动，等待任务执行...")
    # 用事件等待代替手动while循环，让事件循环专注处理异步任务
    await asyncio.Event().wait()

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

2. 替换手动while循环 #

原代码中的while True + await asyncio.sleep(10)虽不会阻塞事件循环，但用await asyncio.Event().wait()能更优雅地让事件循环持续运行，减少不必要的代码逻辑，确保调度器任务能被及时处理。

3. 修正拼写规范 #

原代码中sheduler为拼写错误（正确应为scheduler），虽Python动态特性不会导致报错，但规范拼写能避免后续维护时的混淆。

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