Django中APScheduler定时函数意外执行3次的原因及解决方法

使用Django开发Web应用，通过APScheduler每分钟向数据表中添加1条数据，但在Django管理面板查看时，发现每分钟生成3条数据。相关代码及版本信息如下：

# models.py
from django.db import models
class Test(models.Model):
    test = models.CharField(max_length=8)
    created_at = models.DateTimeField(auto_now_add=True)

# admin.py
from django.contrib import admin
from .models import Test
class TestAdmin(admin.ModelAdmin):
    list_display = ('test', 'created_at')

admin.site.register(Test, TestAdmin)

# ap_scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler
from .models import Test

def myfunc():
    Test.objects.create(test='test')

def start():
    scheduler = BackgroundScheduler()
    scheduler.add_job(myfunc, 'cron', second=50)
    scheduler.start()

# apps.py
from django.apps import AppConfig
class MembersConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'my_app_name'
    
    def ready(self):
        from .ap_scheduler import start
        start()

版本信息 #

APScheduler==3.9.1
Django==4.0.5

原因分析 #

核心问题是Django的ready()方法被多次调用，导致APScheduler被实例化并启动了多个独立调度器，每个调度器都会执行定时任务：

• 使用runserver启动项目时，Django默认会启动两个进程（主进程负责代码重载，子进程处理请求），自动重载机制会额外触发ready()调用；
• 若用多进程WSGI服务器（如Gunicorn多worker），每个worker进程都会执行ready()，各自启动调度器；
• 你的代码每次调用start()都会新建BackgroundScheduler实例，多个实例并行运行就会造成任务重复执行。

解决方法 #

方法1：仅在主进程启动调度器 #

修改apps.py，通过环境变量判断是否为主进程，避免重复启动：

from django.apps import AppConfig
import os

class MembersConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'my_app_name'
    
    def ready(self):
        # RUN_MAIN是Django自动重载时设置的变量，主进程值为true
        if os.environ.get('RUN_MAIN') == 'true':
            from .ap_scheduler import start
            start()

方法2：禁用runserver自动重载（开发环境用） #

启动项目时添加--noreload参数，避免多进程触发重复调度：

python manage.py runserver --noreload

方法3：用文件锁防止重复启动 #

修改ap_scheduler.py，通过文件锁确保同一时间只有一个调度器运行：

from apscheduler.schedulers.background import BackgroundScheduler
from .models import Test
import fcntl
import os

def myfunc():
    Test.objects.create(test='test')

def start():
    lock_file = open('/tmp/django_scheduler.lock', 'w')
    try:
        # 获取排他锁，无法获取则跳过启动
        fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        scheduler = BackgroundScheduler()
        scheduler.add_job(myfunc, 'cron', second=50)
        scheduler.start()
    except BlockingIOError:
        pass

方法4：独立启动调度器进程（生产环境推荐） #

不要将调度器和Web进程绑定，单独启动调度器进程：

1. 创建独立启动脚本scheduler_start.py：

import os
import django

# 配置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
django.setup()

from my_app_name.ap_scheduler import myfunc
from apscheduler.schedulers.background import BackgroundScheduler

if __name__ == '__main__':
    scheduler = BackgroundScheduler()
    scheduler.add_job(myfunc, 'cron', second=50)
    scheduler.start()
    print("Scheduler started...")
    # 保持进程运行
    try:
        while True:
            pass
    except KeyboardInterrupt:
        scheduler.shutdown()

2. 单独启动调度器：

python scheduler_start.py

3. 修改apps.py，移除ready()中的调度器启动代码，Web进程正常启动即可。

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