Celery autodiscover_tasks无法发现Django应用任务的问题排查

问题现象 #

在Django项目视图中执行以下代码时，仅能获取Celery内置任务，无法看到Django应用myapp下定义的自定义任务：

from myproj.celery import app
print(app.tasks)

输出仅包含Celery自身任务：

{'celery.group': <@task: celery.group of myproj at 0x7fa640101940>, 'celery.map': <@task: celery.map of myproj at 0x7fa640101940>, 'celery.chain': <@task: celery.chain of myproj at 0x7fa640101940>, 'celery.backend_cleanup': <@task: celery.backend_cleanup of myproj at 0x7fa640101940>, 'celery.starmap': <@task: celery.starmap of myproj at 0x7fa640101940>, 'celery.chord': <@task: celery.chord of myproj at 0x7fa640101940>, 'celery.accumulate': <@task: celery.accumulate of myproj at 0x7fa640101940>, 'celery.chunks': <@task: celery.chunks of myproj at 0x7fa640101940>, 'celery.chord_unlock': <@task: celery.chord_unlock of myproj at 0x7fa640101940>}

只有直接导入应用的tasks模块后，才能看到自定义任务：

from myapp.tasks import app
print(app.tasks)

项目celery.py配置：

# celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproj.settings')

app = Celery('myproj')
app.conf.update({
    'broker_url': 'redis://localhost:8000/11',
    'result_backend': 'django-db',
    'accept_content': ['json'],
    'task_serializer': 'json',
    'result_serializer': 'json',
    'worker_force_execv': True,
    'worker_concurrency': 10,
    'task_time_limit': 60 * 60,
    'worker_max_tasks_per_child': 200,
    'beat_tz_aware': False,
    'result_extended': True,
})
app.conf.enable_utc = False
app.autodiscover_tasks()

myapp/tasks.py任务代码：

from myproj.celery import app
@app.task
def add():
    x=1
    y=y

原因分析 #

autodiscover_tasks()未生效的核心原因包括：

1. 任务模块存在语法错误：myapp/tasks.py中add函数的y=y属于未定义变量，Celery自动发现时导入模块会触发NameError，导致任务注册失败。
2. 应用未在INSTALLED_APPS中注册：如果myapp没添加到Django配置settings.py的INSTALLED_APPS列表，autodiscover_tasks()默认不会扫描该应用的tasks.py。
3. 自动发现范围未明确指定：项目结构复杂时，仅调用app.autodiscover_tasks()可能无法正确定位到任务模块。

解决步骤 #

1. 修复任务模块语法错误 #

修改myapp/tasks.py代码，确保模块能正常导入：

from myproj.celery import app
@app.task
def add():
    x = 1
    y = 2  # 替换原错误的y=y，或按业务逻辑调整
    return x + y

2. 确认应用已注册到INSTALLED_APPS #

打开myproj/settings.py，确保INSTALLED_APPS包含myapp：

INSTALLED_APPS = [
    # 其他内置/第三方应用
    'myapp',
]

3. 显式指定自动发现的应用范围 #

在celery.py中，将autodiscover_tasks()改为显式传入应用列表：

from django.conf import settings
# 替换原app.autodiscover_tasks()
app.autodiscover_tasks(settings.INSTALLED_APPS)

或仅指定特定应用：

app.autodiscover_tasks(['myapp'])

4. 验证修复效果 #

重启Celery Worker服务，在视图中重新执行以下代码：

from myproj.celery import app
print(app.tasks)

此时输出应包含myapp.tasks.add等自定义任务。

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