Celery autodiscover_tasks 不适用于所有 Django 1.7 应用程序答案

【问题标题】：Celery autodiscover_tasks not working for all Django 1.7 appsCelery autodiscover_tasks 不适用于所有 Django 1.7 应用程序
【发布时间】：2015-04-22 03:00:35
【问题描述】：

我有一个带有 Celery 3.1 的 Django 1.7 项目。我的 Django 项目中的所有应用程序都使用新的 AppConfig。问题是不是所有的任务都用autodiscover_tasks找到：

app.autodiscover_tasks(settings.INSTALLED_APPS)

如果我像这样使用 autodiscover_tasks，它会起作用：

app.autodiscover_tasks(settings.INSTALLED_APPS + ('apps.core','apps.sales'))

找到了网站中定义的任务，但没有找到核心和销售中的任务。所有的布局都与apps.py 和tasks.py 相同。

项目文件夹结构为：

apps
  core
  apps.py
  tasks.py
dashboard
  apps.py
sales
  apps.py
  tasks.py
websites
  apps.py
  tasks.py

类定义如下：

class WebsitesConfig(AppConfig):
    name = 'apps.websites'
    verbose_name = 'Websites'

class SalesConfig(AppConfig):
    name = 'apps.sales'
    verbose_name = 'Sales'

【问题讨论】：

标签： python django celery

【解决方案1】：

这在许多 Celery 问题中都有讨论，例如 #2596 和 #2597。

如果您使用的是 Celery 3.x，修复方法是使用：

from django.apps import apps
app.autodiscover_tasks(lambda: [n.name for n in apps.get_app_configs()])

正如#3341 中提到的，如果您使用的是 Celery 4.x（即将发布），您可以使用：

app.autodiscover_tasks()

【讨论】：

天哪，你拯救了一天！估计 1 小时的工作变成了一整天令人头疼的工作。如果没有你的回答，我今晚会做噩梦。非常感激。干杯。
我用的是django 2.1和celery 4.2，还是有这个问题需要加get_app_configs:(
@TarasMatsyk 这对我来说就是这种情况，但 Celery 3.x 的“变通”仍然对我有用。

【解决方案2】：

我刚刚遇到这个问题是因为虚拟环境配置错误。

如果已安装的应用在运行 celery 的虚拟环境中缺少依赖项，则不会自动发现已安装应用的任务。当我从在同一台机器上运行我的 Web 服务器和 celery 转向分布式解决方案时，这让我很受打击。错误的构建会导致不同节点上的环境文件不同。

我添加了缺少的依赖项，然后重新启动了 celery 服务。

【讨论】：

python -m pip check能查到吗？
我想你可能会这样做；提供requirements.txt 是准确和最新的。当发生此特定更改时，这是在机器上不正确的几个环境文件之一。

【解决方案3】：

我必须将它添加到定义我的 celery 应用程序的模块中：

from __future__ import absolute_import

【讨论】：

【解决方案4】：

我的问题是 celery 命令中 celery.py 的路径错误。命令应该是这样的：

celery worker -A <project_name.celery_app_module> -l info

哪里的芹菜celery_app_module是这样的：

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

# Set default Django settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '<project_name>.settings')

app = Celery('<project_name>', include=['<app_name>.tasks'])
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

【讨论】：