Celery Beat执行Django周期性任务时遇AppRegistryNotReady错误求助

这个错误的核心原因是Celery启动过程中过早导入了依赖Django模型的任务代码，此时Django的应用注册表还没完成加载。结合你的配置，咱们可以通过以下几步修复：

1. 调整Celery初始化文件（celery.py） #

确保Celery在初始化时正确触发Django的应用加载，完整的celery.py应该是这样：

from __future__ import absolute_import, unicode_literals
import os
import django
from celery import Celery

# 设置Django配置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')

# 手动触发Django应用加载，这是关键步骤
django.setup()

app = Celery('myapp')
# 从Django配置中导入Celery设置，使用CELERY_前缀
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现所有Django应用中的tasks.py文件
app.autodiscover_tasks()

2. 延迟任务模块中的模型导入 #

不要在tasks.py的顶部直接导入Django模型，而是把导入操作放到任务函数内部执行。比如你的periodic_task应该改成这样：

from celery import shared_task

@shared_task
def periodic_task():
    # 只在任务执行时才导入模型，此时Django应用已经加载完成
    from myapp.models import YourModel
    # 创建并保存模型的逻辑
    new_instance = YourModel(...)
    new_instance.save()

3. 检查Settings中的Celery配置 #

确保你的settings.py里正确导入了timedelta，并且配置格式正确：

from datetime import timedelta

CELERY_IMPORTS = ('myapp.tasks',)
CELERY_BEAT_SCHEDULE = {
    'periodic_task': {
        'task': 'myapp.tasks.periodic_task',
        'schedule': timedelta(seconds=1),
    },
}

为什么这样能解决问题？ #

Celery在启动worker或beat进程时，会提前导入任务模块。如果任务模块顶部直接引用Django模型，此时Django的应用还没完成初始化，就会抛出AppRegistryNotReady错误。通过django.setup()强制Celery初始化Django应用，再结合延迟导入模型，就能保证任务执行时Django的所有应用都已经加载完毕。

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