### Celery任务模块未导入错误的解决方案 在使用Celery时，如果遇到 `Did you remember to import the module containing this task?` 错误，通常是因为任务模块未被正确导入或注册。以下是可能的原因及解决方法： #### 1. 确保任务模块已正确注册 在 `celery.py` 文件中，确保调用了 `autodiscover_tasks` 方法，并将所有包含任务的模块路径传递给它。例如： ```python from celery import Celery app = Celery('your_project_name') app.config_from_object('django.conf:settings', namespace='CELERY') # 自动发现任务 app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) ``` 上述代码会自动从项目的 `INSTALLED_APPS` 中查找任务模块[^4]。 #### 2. 检查消息队列中的缓存任务 有时，消息队列（如 RabbitMQ 或 Redis）中可能存在旧的任务定义。清理队列并重启 Celery Worker 可以解决问题。例如，对于 Redis 队列，可以执行以下命令清除数据： ```bash redis-cli flushall ``` 然后重启 Celery Worker 和相关服务[^4]。 #### 3. 使用显式导入任务的方式 如果自动发现任务不可行，可以通过手动配置任务模块来确保任务被加载。在 Django 项目中，可以在 `settings.py` 中添加如下配置： ```python CELERY_IMPORTS = ( 'your_app.tasks', 'another_app.tasks', ) ``` 这将显式地导入指定的任务模块。 #### 4. 避免相对导入问题 相对导入可能导致任务名称不匹配，从而引发 `NotRegistered` 异常。建议为每个任务明确指定名称。例如： ```python from celery import shared_task @shared_task(name='your_app.tasks.add') def add(x, y): return x + y ``` 通过这种方式，可以避免因相对导入导致的任务名称冲突[^5]。 #### 5. 检查任务是否被正确调用 确保任务在调用时使用了正确的名称。例如，如果任务定义为： ```python @app.task(name='proj.tasks.add') def add(x, y): return x + y ``` 则在调用时应使用相同的名称： ```python from proj.tasks import add add.delay(1, 2) ``` #### 6. 调试任务注册情况 可以通过打印 Celery 的任务列表来确认任务是否已被注册。运行以下命令查看当前注册的任务： ```bash celery -A your_project_name inspect registered ``` 如果目标任务未出现在输出中，则需要检查任务模块的导入逻辑[^3]。 --- ### 示例代码 以下是一个完整的 Celery 配置示例，展示了如何避免常见的任务导入问题： ```python # celery.py from __future__ import absolute_import, unicode_literals import os from celery import Celery os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings') app = Celery('your_project') app.config_from_object('django.conf:settings', namespace='CELERY') app.autodiscover_tasks() # tasks.py from celery import shared_task @shared_task(name='your_app.tasks.add') def add(x, y): return x + y ``` ---