如何实现Django应用启动后仅执行一次依赖ORM的特定函数？

这确实是Django部署里挺头疼的一个场景——既要依赖ORM就绪，又要避免多worker/多容器重复执行，还不能被管理命令触发。我来分享几个经过实践验证的方案，完美匹配你的需求：

方案一：改进AppConfig.ready()，过滤命令+分布式锁 #

这个方案是对原生ready()方法的优化，解决了管理命令误触发和多worker重复执行的问题，步骤如下：

1. 先创建一个锁模型（用于分布式锁） #

在你的app里新建一个模型，用来标记初始化任务是否已经执行：

# your_app/models.py
from django.db import models

class StartupLock(models.Model):
    key = models.CharField(max_length=64, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name = "启动锁"
        verbose_name_plural = "启动锁"

记得先执行makemigrations和migrate创建这个表。

2. 修改AppConfig的ready()方法 #

在apps.py里，我们先判断当前执行的是不是服务器启动命令（比如runserver、gunicorn），不是的话直接跳过；然后用锁确保只有第一个进程执行初始化任务：

# your_app/apps.py
import sys
from django.apps import AppConfig
from django.utils import timezone

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

    def ready(self):
        # 第一步：过滤非服务器启动命令
        # 匹配runserver或gunicorn启动命令
        is_server_command = any(cmd in sys.argv for cmd in ['runserver', 'gunicorn'])
        # 明确排除常见管理命令，避免误触发
        management_commands = ['makemigrations', 'migrate', 'shell', 'createsuperuser']
        if any(cmd in sys.argv for cmd in management_commands):
            return
        
        if not is_server_command:
            return

        # 第二步：用分布式锁确保仅执行一次
        from your_app.models import StartupLock

        try:
            # 检查锁是否存在，且未过期（1小时过期，避免异常残留锁）
            lock, created = StartupLock.objects.get_or_create(key='initialization_task')
            if not created:
                one_hour_ago = timezone.now() - timezone.timedelta(hours=1)
                if lock.created_at < one_hour_ago:
                    lock.delete()
                    StartupLock.objects.create(key='initialization_task')
                else:
                    return  # 锁有效，已执行过任务
            
            # --------------------------
            # 这里写你的初始化代码（依赖ORM的逻辑）
            print("开始执行一次性初始化任务...")
            # 示例：批量导入数据
            # from your_app.models import YourModel
            # YourModel.objects.bulk_create([...])
            # --------------------------

        except Exception as e:
            print(f"初始化任务执行失败：{str(e)}")
            # 可选：删除锁，下次启动重试
            StartupLock.objects.filter(key='initialization_task').delete()

方案优势 #

• 兼容本地开发（runserver）和生产部署（gunicorn多worker）
• 完全避开makemigrations、shell等管理命令的触发
• 数据库锁无需额外依赖，多容器部署时只要共享同一个数据库，就能保证整个站点仅执行一次
• 过期锁逻辑避免了进程异常退出导致的任务无法重新执行

方案二：环境变量控制+锁（更灵活的开发体验） #

如果想让本地开发人员更清晰地控制是否执行初始化任务，可以结合环境变量：

1. 启动命令中添加环境变量 #

• 本地开发启动时：

RUN_STARTUP_TASK=true python manage.py runserver

• Docker容器启动时（在docker-compose.yml中设置）：

# docker-compose.yml
services:
  web:
    build: .
    environment:
      - RUN_STARTUP_TASK=true
    command: gunicorn your_project.wsgi:application

2. 修改ready()方法判断环境变量 #

在apps.py的ready()开头加上：

import os
# ...
def ready(self):
    if os.environ.get('RUN_STARTUP_TASK') != 'true':
        return
    # 后续锁逻辑和方案一一致

方案优势 #

• 开发人员可以明确控制是否执行任务，避免本地调试时意外触发耗时任务
• Docker部署时自动设置环境变量，无需手动操作

注意事项 #

• 如果使用gunicorn的preload_app=True配置，ready()会在fork worker前执行，但锁逻辑依然能保证仅执行一次，不用担心重复问题
• 初始化任务如果耗时很长，会导致第一个worker启动变慢，但不影响其他worker的创建
• 如果你用Redis作为缓存，可以替换数据库锁为Redis锁（比如用redis-py的setnx命令），性能更好，适合高并发场景

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