如何确保Django Channels启动时代码仅执行一次（兼容多Worker）

我之前在做Django Channels项目时也踩过这个坑！AppConfig.ready在多worker环境下（不管是runserver自动重载还是daphne多进程）确实会被每个进程执行一遍，导致初始化代码重复运行——比如我当时注册的定时任务多跑了好几个实例，排查了半天才找到原因。下面分享几个亲测有效的方案，既能兼容开发服务器，又能适配daphne多worker场景：

方案1：本地文件锁（适合单服务器/开发环境） #

利用文件排他锁的特性，确保只有第一个启动的进程能执行初始化代码，适合开发环境或者单机器部署的生产场景。

from django.apps import AppConfig
import fcntl
import os
import sys

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        # 跳过runserver自动重载的子进程（开发环境专属）
        if 'runserver' in sys.argv and 'reload' in sys.argv:
            return

        # 定义锁文件路径，建议放在项目根目录或系统临时目录
        lock_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'app_init.lock')
        lock_file = open(lock_file_path, 'w')
        
        try:
            # 尝试获取非阻塞排他锁
            fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
            # --------------------------
            # 这里放你需要只执行一次的代码
            print("初始化操作执行成功，仅运行一次！")
            # --------------------------
        except BlockingIOError:
            # 其他进程获取锁失败，直接跳过
            pass
        finally:
            # 无需手动关闭文件，进程退出时会自动释放锁
            pass

方案2：Redis分布式锁（适合多服务器/容器化部署） #

如果是分布式环境（多机器、Docker集群），文件锁就失效了，这时候用Redis的原子操作实现分布式锁，能跨机器确保只有一个进程执行初始化。

首先安装依赖：pip install redis

from django.apps import AppConfig
import redis
import sys
from django.conf import settings

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        # 跳过runserver重载进程
        if 'runserver' in sys.argv and 'reload' in sys.argv:
            return

        # 连接Redis（配置从settings里取）
        r = redis.Redis(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            db=settings.REDIS_DB,
            password=settings.REDIS_PASSWORD
        )
        lock_key = "myapp_global_init_lock"
        
        # 设置锁：nx=True表示仅当key不存在时才设置成功，ex=30是锁的过期时间（防止进程意外退出锁未释放）
        lock_acquired = r.set(lock_key, "locked", nx=True, ex=30)
        
        if lock_acquired:
            try:
                # --------------------------
                # 这里放分布式环境下的初始化代码
                print("分布式环境初始化完成，仅运行一次！")
                # --------------------------
            finally:
                # 执行完后手动释放锁
                r.delete(lock_key)
        else:
            # 其他进程获取锁失败，跳过执行
            pass

方案3：Channels Worker ID标识（仅Channels环境可用） #

Channels启动worker时会给每个进程设置CHANNEL_WORKER_ID环境变量，第一个worker的ID是0，我们可以只让ID为0的进程执行初始化。

from django.apps import AppConfig
import os
import sys

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        # 跳过runserver重载进程
        if 'runserver' in sys.argv and 'reload' in sys.argv:
            return

        # 仅让第一个Channels Worker执行初始化
        worker_id = os.environ.get('CHANNEL_WORKER_ID', '0')
        if worker_id == '0':
            # --------------------------
            # 这里放初始化代码
            print("仅第一个Channels Worker执行初始化！")
            # --------------------------

额外建议 #

不管用哪种方案，都尽量把初始化代码做成幂等的——也就是重复执行也不会产生副作用（比如重复创建数据表、重复注册定时任务），这样即使偶尔出现多次执行的情况，也不会导致系统故障。

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