Flask + APScheduler 在多进程下重复触发任务，因每个 worker 独立初始化 scheduler 且无分布式锁；解决法：单 worker 部署、Redis 锁保障互斥、正确管理生命周期。
为什么Flask + APScheduler 在多进程下会重复触发任务
Flask 默认启用多进程（比如 debug=False 时用 gunicorn 或 uwsgi），每个 worker 进程都会独立初始化一个 APScheduler 实例，导致同一任务被多个进程同时执行。这不是 bug，而是设计使然——APScheduler 本身无分布式锁机制。
常见现象包括：日志里看到同一任务每分钟打两行、数据库写入重复记录、邮件/消息被发多次。
- 检查是否启用了多 worker：
gunicorn --workers 4或uwsgi --processes 2 - 确认
APScheduler初始化代码没放在if __name__ == '__main__':下，否则在 reload 模式下也可能重复注册 - Flask 开发服务器（
app.run()）单进程默认不触发该问题，但上线后几乎必现
用 BackgroundScheduler + 单进程守护是最简解法
如果部署环境允许只用单 worker（比如轻量内部服务），直接禁用多进程，让调度器只在一个进程中运行即可。这是最不容易出错的路径。
- gunicorn 启动时加
--workers 1 --threads 4，避免 fork 多份调度器 - 确保
BackgroundScheduler实例全局唯一，且在 Flask app 创建后、启动前初始化 - 不要在路由函数或请求上下文中 start scheduler；应在模块顶层或
create_app()结束后调用scheduler.start()
示例关键片段：
立即学习“Python免费学习笔记（深入）”；
Python 环境工具专业版功能概述Python 环境工具专业版是一项面向实际任务的技能。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。核心要点使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要求处理异常情况。该技能适合需要稳定复用相关能力的场景，可作为自动化工作流的一部分，也便于后续检查、调整和扩展。执行完成后建议检查结果是否符合预期，并保留必要记录。使用与执行同时注意输入完整性、执行条件和输出可用性。从功能定位来看，该技能
from apscheduler.schedulers.background import BackgroundScheduler <p>scheduler = BackgroundScheduler() scheduler.add_job(func=your_task, trigger='interval', minutes=5)</p><h1>必须在 app.run() 或 gunicorn 加载完成后调用</h1><p>if not scheduler.running: scheduler.start()
需要多 worker？必须引入外部锁机制
真正生产环境往往要求高可用和横向扩展，这时就得靠外部协调——推荐用 Redis 锁，比文件锁、数据库行锁更可靠且无单点故障风险。
- 用
redis-py的lock()方法包裹任务执行逻辑，不是锁整个 job，而是锁「本次执行」 - 锁 key 建议含任务标识 + 时间戳（如
f"job:cleanup:{int(time.time() // 300)}"），避免长任务锁死 - 务必设
timeout和blocking_timeout，防止死锁；超时值应略大于任务最大预期耗时
片段示意：
import redis
r = redis.Redis()
def guarded_task():
lock_key = "job:sync_users"
with r.lock(lock_key, timeout=300, blocking_timeout=5):
# 此处放实际业务逻辑
sync_users_to_external_api()
别忽略 Flask 应用生命周期与 scheduler 的耦合风险
Flask 扩展（如 flask-apscheduler）封装了自动管理，但容易掩盖底层问题：比如应用热重载时 scheduler 被反复 init/start，或 SIGTERM 未正确 shutdown 导致残留线程。
- 手动管理时，务必在
atexit.register(scheduler.shutdown)或 Flask 的@app.teardown_appcontext中调用shutdown(wait=False) - 若用
flask-apscheduler，确认配置项SCHEDULER_JOBSTORES没误配成内存型（dict），否则重启后任务丢失 - 日志中出现
Job took longer than ... seconds提示，说明任务阻塞了 scheduler 线程池，应改用ThreadPoolExecutor异步委托，而非在 job 内直接跑耗时操作
真正麻烦的从来不是“怎么加定时任务”，而是“怎么让它只跑一次、跑得稳、挂了能恢复”。锁、单点、生命周期这三块，漏掉任何一块都可能让定时任务从省心功能变成线上火警。
