Flask搭配uwsgi+nginx部署时APScheduler get_jobs()返回空任务列表

核心问题根源 #

• 进程内存空间完全隔离：你在webapp/__init__.py里初始化的GeventScheduler实例，运行在uWSGI的worker进程上下文；而@timer(target='spooler')装饰的check_run_tasks，运行在uWSGI独立的spooler进程中。两个进程的内存完全不互通，哪怕你在tasks.py里导入webapp.scheduler，拿到的也是spooler进程加载webapp模块时新生成的、没有添加任何job的空实例，自然返回空列表。
• 代码疏漏：tasks.py里的_scheduler_method既没有导入也没有定义，当前能运行大概率是你本地补了导入但没意识到跨进程的问题。
• 额外隐藏问题：你把scheduler初始化和启动的代码写在webapp全局初始化流程里，uWSGI master fork出的每一个worker都会执行这段代码，导致每个worker都有独立的scheduler实例，定时任务publish_async会被重复执行N次（N=worker进程数）。

可选解决方案 #

方案一：放弃uwsgidecorators定时器，统一用APScheduler调度 #

APScheduler本身就支持间隔触发的定时任务，完全没必要叠加uWSGI spooler的定时器，从根源上避免跨进程问题：

1. 把tasks.py里的检查任务直接封装成普通函数，不需要@timer装饰：

# webapp/tasks.py
from webapp import scheduler

def check_run_tasks():
    # 直接用同一个scheduler实例，天然能拿到所有已添加的job
    job_list = scheduler.get_jobs()
    print(job_list)

2. 把这个检查任务也添加到同一个scheduler实例里，修改webapp/__init__.py的scheduler初始化部分：

# webapp/__init__.py 新增导入
from webapp.tasks import check_run_tasks

# 原来的scheduler.add_job(...)后面追加
scheduler.add_job(check_run_tasks, id="check_jobs", trigger='interval', seconds=10)

3. 解决多worker重复启动scheduler的问题：用uWSGI的postfork钩子，仅在指定worker中启动scheduler，避免每个worker都启动：
   在webapp/__init__.py里加：

import uwsgi

# 注释掉原来全局直接执行的 scheduler.start()、scheduler.add_job 代码
# 仅在编号为0的worker里启动scheduler，其他worker跳过
@uwsgi.postfork_hooks()
def start_scheduler_only_once():
    if uwsgi.worker_id() == 0:
        scheduler.start()
        scheduler.add_job(publish_async, args=["Hello"], id="job", trigger='interval', seconds=2)
        scheduler.add_job(check_run_tasks, id="check_jobs", trigger='interval', seconds=10)

方案二：如果必须保留uWSGI spooler定时器，用持久化存储同步scheduler状态 #

如果业务要求必须用spooler跑检查任务，就不要跨进程访问内存中的scheduler实例，而是把job状态持久化到公共存储里：

1. 修改webapp/__init__.py的scheduler配置，用SQLAlchemyJobStore把job存在数据库中：

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

# 替换原来的scheduler初始化代码
jobstores = {
    'default': SQLAlchemyJobStore(url=app.config['SQLALCHEMY_DATABASE_URI'])
}
scheduler = GeventScheduler(jobstores=jobstores)

2. tasks.py里的检查任务直接查询数据库即可，不需要访问scheduler实例：

# webapp/tasks.py
from uwsgidecorators import timer
from webapp.models import db  # 用你已有的SQLAlchemy实例直接查job表

@timer(10, target='spooler')
def check_run_tasks(args):
    # APScheduler的SQLAlchemyJobStore默认表名是apscheduler_jobs
    job_list = db.session.execute("SELECT id FROM apscheduler_jobs").fetchall()
    print(job_list)

内容的提问来源于stack exchange，提问作者Яша Проценко