django_apscheduler任务重复执行问题：如何在启动时彻底清除调度器状态？

我帮你分析一下问题的核心原因，再给你针对性的解决方案：

问题根源 #

你的函数被多次调用，主要是因为这两个关键点：

1. 多个调度器实例同时运行：Django的ready()方法在启动场景中（比如runserver自动重载、多进程启动）会被触发多次，每次调用你的scheduler.start()都会新建一个BackgroundScheduler实例。这些实例共享同一个DjangoJobStore（数据库存储），导致多个实例同时执行相同任务。
2. 旧进程残留与任务同步：虽然你调用了remove_all_jobs()，但这个操作仅删除当前实例连接的JobStore中的任务，之前启动的旧调度器实例可能还在后台运行。当新任务添加后，所有存活的实例都会同步执行该任务，最终造成函数被重复调用。

解决方案 #

方案1：改用单例调度器，确保仅一个实例运行

修改你的scheduler.py，把调度器改成模块级单例，从根源避免重复创建实例：

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore
from django_apscheduler.models import DjangoJobExecution
import sys

# 模块级单例调度器，全局仅维护一个实例
_scheduler = None

def scheduled_function():
    print("my function is running as scheduled")

def start():
    global _scheduler
    # 先检查是否已有运行中的调度器，避免重复启动
    if _scheduler and _scheduler.running:
        print("调度器已在运行，跳过启动操作")
        return

    # 初始化调度器并连接JobStore
    _scheduler = BackgroundScheduler()
    _scheduler.add_jobstore(DjangoJobStore(), "default")

    # 彻底清理：删除JobStore所有任务 + 清空旧的执行记录
    _scheduler.remove_all_jobs(jobstore='default')
    DjangoJobExecution.objects.all().delete()

    # 添加任务时开启replace_existing，确保同名任务唯一
    _scheduler.add_job(
        scheduled_function,
        'interval',
        seconds=10,
        name='scheduled_function',
        jobstore='default',
        replace_existing=True
    )

    # 安全启动调度器，捕获异常避免崩溃
    try:
        _scheduler.start()
        print("调度器启动成功")
        _scheduler.print_jobs()
    except Exception as e:
        print(f"调度器启动失败：{str(e)}", file=sys.stderr)
        _scheduler.shutdown()
        _scheduler = None

def shutdown():
    """优雅关闭调度器的辅助函数，用于程序退出时清理"""
    global _scheduler
    if _scheduler and _scheduler.running:
        _scheduler.shutdown()
        _scheduler = None

方案2：优化ready()方法的进程判断逻辑

修改apps.py，用更可靠的方式确保只有主进程才启动调度器，避免多场景下的重复触发：

from django.apps import AppConfig
import os
import sys

class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'my_server_app'

    def ready(self):
        # 先排除非Web服务启动场景（比如migrate、shell等管理命令）
        valid_commands = ['runserver', 'runserver_plus', 'gunicorn', 'uwsgi']
        if not any(cmd in sys.argv for cmd in valid_commands):
            return

        # 针对不同部署场景判断是否为主进程
        is_main_process = False
        # 本地开发用runserver时，依赖Werkzeug或Django的主进程标记
        if 'runserver' in sys.argv:
            is_main_process = os.environ.get("RUN_MAIN") == "true" or os.environ.get("WERKZEUG_RUN_MAIN") == "true"
        # 生产环境用gunicorn时，只让第一个worker启动调度器
        elif 'gunicorn' in sys.argv:
            is_main_process = int(os.environ.get("WORKER_ID", 1)) == 1
        # 生产环境用uwsgi时，仅让master进程启动调度器
        elif 'uwsgi' in sys.argv:
            is_main_process = os.environ.get("UWSGI_ROLE") == 'master'

        if is_main_process:
            from my_app.scheduler import start, shutdown
            # 注册退出信号，确保程序关闭时优雅停止调度器
            import atexit
            atexit.register(shutdown)
            start()

方案3：生产环境额外提示

如果是线上部署，还要注意：

• 避免用runserver，改用gunicorn/uwsgi等生产级服务器
• 若使用多worker，要么仅让第一个worker启动调度器，要么用Redis等分布式锁保证仅一个实例运行
• 定期清理DjangoJobExecution表的旧记录，避免数据库冗余

这样修改后，就能彻底解决任务重复执行的问题，确保只有一个调度器实例在运行，并且每次启动都会彻底清理旧任务。

备注：内容来源于stack exchange，提问作者Omroth