Django内部实现定时任务执行方案咨询

方案1：后台线程 + APScheduler（无额外进程） #

APScheduler可以直接嵌入Django进程，无需单独启动命令，只需在App初始化时启动后台线程运行调度器即可。

实现步骤： #

1. 安装依赖：

pip install apscheduler

2. 在目标App下创建tasks.py，编写数据获取逻辑：

import requests
from django.conf import settings

def fetch_daily_data():
    # 调用外部接口获取JSON数据
    response = requests.get(settings.EXTERNAL_DATA_URL)
    raw_data = response.json()
    # 后续数据解析、入库等操作
    # ...

3. 修改App的apps.py，重写ready()方法启动调度器：

from django.apps import AppConfig
import threading
import os
from apscheduler.schedulers.background import BackgroundScheduler
from .tasks import fetch_daily_data

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name'

    def ready(self):
        # 规避runserver启动时的重复进程问题
        if os.environ.get('RUN_MAIN') == 'true':
            scheduler = BackgroundScheduler()
            # 配置每日凌晨2点执行任务
            scheduler.add_job(fetch_daily_data, 'cron', hour=2)
            scheduler.start()

4. 在settings.py中指定使用自定义的App配置：

INSTALLED_APPS = [
    # ...
    'your_app_name.apps.YourAppConfig',
    # ...
]

注意事项： #

• 调度器与Django runserver进程同属一个进程，无需额外启动命令。
• 后续扩容多服务器时，需通过数据库锁/Redis分布式锁避免任务重复执行：比如任务启动前尝试创建当日唯一的执行记录，仅创建成功的实例执行任务。

方案2：原生线程循环（无第三方依赖） #

如果不想引入外部库，可以自行编写后台线程，通过时间循环实现定时逻辑。

实现步骤： #

1. 同样在tasks.py中定义数据获取任务。

2. 在apps.py的ready()方法中启动后台线程：

from django.apps import AppConfig
import threading
import time
import os
from datetime import datetime, timedelta
from .tasks import fetch_daily_data

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name'

    def ready(self):
        if os.environ.get('RUN_MAIN') == 'true':
            def run_scheduler():
                while True:
                    now = datetime.now()
                    # 计算下一次执行时间（次日凌晨2点）
                    next_run = now.replace(hour=2, minute=0, second=0, microsecond=0)
                    if next_run <= now:
                        next_run += timedelta(days=1)
                    # 等待到执行时间
                    time.sleep((next_run - now).total_seconds())
                    # 执行任务
                    fetch_daily_data()

            # 启动守护线程
            thread = threading.Thread(target=run_scheduler, daemon=True)
            thread.start()

注意事项： #

• 完全基于Python原生能力，无需额外依赖，但需自行处理时间计算和循环逻辑。
• 多服务器场景下同样需要加锁机制避免重复执行。

多服务器扩容后的锁实现思路 #

• 数据库锁：任务启动时，尝试通过get_or_create创建一条带当日日期的唯一记录，仅创建成功的实例执行任务。
• Redis分布式锁：利用Redis的SETNX命令获取锁并设置过期时间，只有拿到锁的实例才能执行任务。

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