如何在Telegram Bot重启后恢复Python的AsyncIOScheduler任务？

你的问题核心出在两个地方：

• 调度器启动时机错误，只有用户发/start才触发启动，程序重启后没人触发该命令的话，调度器根本不会运行，自然无法加载数据库里的任务
• 任务持久化的细节没处理到位，缺少必要的配置保证任务能被正确序列化、反序列化

下面直接给你改好的代码并说明关键改动：

关键调整点 #

1. 提前启动调度器：把调度器启动逻辑移到程序初始化阶段，而非用户消息处理器内，确保程序一启动就自动加载数据库中保存的任务
2. 完善任务配置：添加replace_existing=True避免重复创建同ID任务，同时确保任务函数全局可访问，让调度器重启后能找到它
3. 适配异步环境：给调度器添加wrap_non_async_jobs=True配置，避免同步任务阻塞异步事件循环

完整修改代码 #

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from aiogram import Router, Bot, Dispatcher
from aiogram.types import Message
from aiogram.filters import CommandStart
import datetime
import asyncio

# 全局定义任务函数，确保重启后调度器能找到它
def f():
    print("Hello, World!")

# 配置持久化任务存储
jobstores = {
    "default": SQLAlchemyJobStore(url="sqlite:///scheduler.db")
}

# 初始化异步调度器，配置适配非异步任务
scheduler = AsyncIOScheduler(
    jobstores=jobstores,
    job_defaults={"coalesce": False, "max_instances": 1},
    wrap_non_async_jobs=True
)

router = Router()

@router.message(CommandStart())
async def start_handler(message: Message):
    # 添加任务，replace_existing避免重复创建同ID任务
    scheduler.add_job(
        f, 
        "date", 
        id="job_1", 
        jobstore="default", 
        run_date=datetime.datetime.now() + datetime.timedelta(seconds=30),
        replace_existing=True
    )
    await message.reply("任务已添加，程序重启后会自动执行")

async def main():
    # 替换成你的Bot Token
    bot = Bot(token="YOUR_BOT_TOKEN")
    dp = Dispatcher()
    dp.include_router(router)
    
    # 程序启动时立即启动调度器，自动加载数据库中的任务
    scheduler.start()
    
    try:
        await dp.start_polling(bot)
    finally:
        await bot.session.close()
        # 关闭程序时确保任务都保存到数据库
        scheduler.shutdown()

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

额外注意事项 #

• 如果你的任务是异步函数（比如需要操作数据库、调用Telegram API），直接定义成async def f():即可，无需wrap_non_async_jobs配置
• 任务函数必须放在全局作用域，不能定义在某个函数内部，否则调度器重启后会找不到该函数导致报错
• 程序关闭时一定要调用scheduler.shutdown()，确保所有未执行的任务都被持久化到数据库

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