flask中不可直接用threading.timer实现定时任务,因其在开发模式下会被重载机制杀死、生产环境多worker会导致重复执行;应使用flask-apscheduler并注意避reloader二次初始化、job函数需可序列化且避免上下文依赖、生产环境必须切换为sqlalchemyjobstore等持久化存储并加分布式锁。

Flask 里直接用 threading.Timer 不可靠
很多新手会想:我手动开个 threading.Timer 或者循环 time.sleep() 不就能跑定时任务了吗?不行。Flask 开发服务器(flask run)默认启用重载(reloader),每次代码变动都会重启进程,所有后台线程直接被杀掉;生产环境用 Gunicorn/Uvicorn 时,多 worker 模式还会导致任务重复执行。定时任务必须交由能感知应用生命周期、支持持久化和去重的调度器来管。
Flask-APScheduler 初始化要避开 reloader 二次加载
Flask-APScheduler 底层基于 APScheduler,但它的 Scheduler 实例不能在模块顶层创建,否则开发模式下会被初始化两次——一次是主进程,一次是 reloader 子进程,后者会报 Job already added 或直接崩溃。
- 把调度器实例化放在
create_app()工厂函数内部,且加if app.debug and os.environ.get("WERKZEUG_RUN_MAIN") != "true": return守卫 - 或者更稳妥:只在
app.app_context()中启动,且用scheduler.start(paused=True),等所有配置加载完再scheduler.resume() - 别把 job 函数写在路由文件里——它可能被多次 import,造成重复注册
job 函数必须可序列化,且避免闭包引用 Flask 上下文
APScheduler 默认使用 MemoryJobStore,但 job 对象需要被 pickle 序列化(比如你换用 SQLAlchemyJobStore 或部署多实例时)。这意味着:
- job 函数必须是模块顶层函数,不能是类方法、lambda 或嵌套函数
- 不能直接在 job 里调用
current_app或request——此时没有请求上下文,会抛RuntimeError: Working outside of application context - 正确做法:在 job 函数开头手动推入应用上下文:
with app.app_context():<br> db.session.query(...).update(...)<br> db.session.commit()
生产环境务必换用持久化 JobStore 并禁用内存模式
默认的 MemoryJobStore 只存内存,进程一挂任务全丢,而且多 worker 下各干各的,极易重复执行。上线前必须改:
- 用
SQLAlchemyJobStore:装sqlalchemy,配置SQLALCHEMY_DATABASE_URI,然后传给APScheduler的jobstores参数 - 确保数据库表已建好:
python -c "from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore; SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')._create_tables()" - 禁用 debug 模式下的 reloader:
flask run --no-reload,或 Gunicorn 启动时设--preload避免每个 worker 单独初始化调度器
跨机器部署时,还得考虑分布式锁——APScheduler 本身不提供,得靠数据库行锁或 Redis SETNX 自己兜底。这点容易被忽略,但一旦任务逻辑涉及资金、库存等敏感操作,后果就是双倍扣减。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











