Flask集成APScheduler次日调度失效问题排查求助

问题现象 #

• 注册调度任务后，首次执行正常
• 次日调度任务完全无输出（打印、日志均无记录）
• Gunicorn 输出 WORKER TIMEOUT 日志并重启应用，之后间歇性出现 schedule was missed by 00:00 日志
• 当前Gunicorn worker数量配置为1

相关代码片段 #

my_app/setup.py #

# my_app/setup.py
app = None
sched = None

class customFlask(Flask):
    """
    Flask Setting
    """

    def __init__(self, **kwargs):
        Flask.__init__(self, __name__, **kwargs)
        base.app = self
        self.config.from_envvar("SETTINGS")

        base.db = _BaseSQLAlchemy(base.app, session_options={"autoflush": False})
        Migrate(base.app, base.db)

def init(**kwargs):
    """
    Setting Flask (call CorFlask)
    """
    global app, sched
    if not app:
        app = customFlask(**kwargs)
    sched = BackgroundScheduler(
        jobstores={"default": SQLAlchemyJobStore(url=SQLALCHEMY_DATABASE_URI, tablename="apscheduler_jobs")}
    )


    return app, sched

my_app/init.py #

# my_app/__init__.py
app, sched = init(instance_path=os.path.dirname(__file__) + "/../instance")
core_bp = Blueprint("core", "core_bp", template_folder=os.path.dirname(__file__) + "/templates")

with app.app_context():
    # (...skip...)

    sched.start()

初步怀疑点 #

• BackgroundScheduler 在 Flask app_context 上下文内启动，可能导致上下文过期或被回收
• 仅配置1个Gunicorn worker，worker超时重启后，原调度器进程被销毁，新启动的worker未重新加载调度任务

问题分析与解决方案 #

核心问题拆解 #

Gunicorn的worker是独立进程，单个worker因超时重启时，原进程内的BackgroundScheduler会被销毁。当前代码中调度器启动在单次app_context包裹内，重启后的新worker不会自动从SQLAlchemyJobStore加载已有任务，导致后续调度完全失效。

具体修复步骤 #

1. 调整调度器启动时机

将sched.start()移到app_context外部，确保worker启动时能自动从jobstore加载已存储的任务：

# my_app/__init__.py
app, sched = init(instance_path=os.path.dirname(__file__) + "/../instance")
core_bp = Blueprint("core", "core_bp", template_folder=os.path.dirname(__file__) + "/templates")

with app.app_context():
    # (...skip 原有初始化逻辑...)

# 启动调度器，自动加载jobstore中的任务
sched.start()

2. 优化Gunicorn配置

• 增加worker数量至2及以上（例如启动参数--workers=2），单个worker重启时，其他worker的调度器仍能正常运行，避免任务中断
• 根据任务实际执行时长调整超时参数（例如--timeout=300），减少无意义的worker重启

3. 确保任务执行时的上下文可用

如果任务函数需要访问Flask上下文（如数据库、配置），需在任务内部手动推入上下文：

def scheduled_task():
    with app.app_context():
        # 任务逻辑，例如数据库操作、日志输出
        print("任务执行中")

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