Celery收到未注册的类型任务(运行示例)

Question

我正试图从Celery文档中运行示例.

我跑: celeryd --loglevel=INFO

/usr/local/lib/python2.7/dist-packages/celery/loaders/default.py:64: NotConfigured: No 'celeryconfig' module found! Please make sure it exists and is available to Python.
  "is available to Python." % (configname, )))
[2012-03-19 04:26:34,899: WARNING/MainProcess]  

 -------------- celery@ubuntu v2.5.1
---- **** -----
--- * ***  * -- [Configuration]
-- * - **** ---   . broker:      amqp://guest@localhost:5672//
- ** ----------   . loader:      celery.loaders.default.Loader
- ** ----------   . logfile:     [stderr]@INFO
- ** ----------   . concurrency: 4
- ** ----------   . events:      OFF
- *** --- * ---   . beat:        OFF
-- ******* ----
--- ***** ----- [Queues]
 --------------   . celery:      exchange:celery (direct) binding:celery

tasks.py:

# -*- coding: utf-8 -*-
from celery.task import task

@task
def add(x, y):
    return x + y

run_task.py:

# -*- coding: utf-8 -*-
from tasks import add
result = add.delay(4, 4)
print (result)
print (result.ready())
print (result.get())

在同一文件夹celeryconfig.py中:

CELERY_IMPORTS = ("tasks", )
CELERY_RESULT_BACKEND = "amqp"
BROKER_URL = "amqp://guest:guest@localhost:5672//"
CELERY_TASK_RESULT_EXPIRES = 300

当我运行"run_task.py"时:

在python控制台上

eb503f77-b5fc-44e2-ac0b-91ce6ddbf153
False

celeryd服务器上的错误

[2012-03-19 04:34:14,913: ERROR/MainProcess] Received unregistered task of type 'tasks.add'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you are using relative imports?
Please see http://bit.ly/gLye1c for more information.

The full contents of the message body was:
{'retries': 0, 'task': 'tasks.add', 'utc': False, 'args': (4, 4), 'expires': None, 'eta': None, 'kwargs': {}, 'id': '841bc21f-8124-436b-92f1-e3b62cafdfe7'}

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/consumer.py", line 444, in receive_message
    self.strategies[name](message, body, message.ack_log_error)
KeyError: 'tasks.add'

请解释一下是什么问题.

Answer 1

我认为你需要重启工作服务器.我遇到了同样的问题并通过重新启动来解决它.

Answer 2

我遇到了同样的问题:原因"Received unregistered task of type.."是celeryd服务没有找到并在服务启动时注册任务(顺便说一句,当你开始时他们的列表是可见的 ./manage.py celeryd --loglevel=info).

应CELERY_IMPORTS = ("tasks", )在设置文件中声明这些任务.
如果你有一个特殊的celery_settings.py文件,它必须在celeryd服务启动时声明,--settings=celery_settings.py就像digivampire写的那样.

Answer 3

您可以在celery.registry.TaskRegistry班级中查看已注册任务的当前列表.可能是你的celeryconfig(在当前目录中)不在,PYTHONPATH所以芹菜找不到它并回退到默认值.只需在启动芹菜时明确指定.

celeryd --loglevel=INFO --settings=celeryconfig

你也可以设置--loglevel=DEBUG,你应该立即看到问题.

Answer 4

无论您使用CELERY_IMPORTS或autodiscover_tasks,重要的一点是任务能够被发现的芹菜登记任务的名称应与工人试图取的名字.

例如,当您启动Celery时celery worker -A project --loglevel=DEBUG,您应该看到任务的名称.例如,如果我debug_task在我的任务中有任务celery.py.

[tasks]
. project.celery.debug_task
. celery.backend_cleanup
. celery.chain
. celery.chord
. celery.chord_unlock
. celery.chunks
. celery.group
. celery.map
. celery.starmap

如果不能在列表中看到您的任务,请检查您的芹菜配置正确导入的任务,无论是在--setting,--config,celeryconfig或config_from_object.

如果您使用的是celery beat,请确保task您使用的任务名称CELERYBEAT_SCHEDULE与芹菜任务列表中的名称相匹配.

Answer 5

对我有用的是向 celery 任务装饰器添加明确的名称。我将任务声明从 更改@app.tasks为@app.tasks(name='module.submodule.task')

这是一个例子

起初我的任务是这样的：

# tasks/test_tasks.py
@celery.task
def test_task():
    print("Celery Task  !!!!")

我把它改为：

# tasks/test_tasks.py
@celery.task(name='tasks.test_tasks.test_task')
def test_task():
    print("Celery Task  !!!!")

当您没有专用的tasks.py 文件将其包含在celery 配置中时，此方法很有用。

Answer 6

我也有同样的问题; 我补充道

CELERY_IMPORTS=("mytasks")

在我的celeryconfig.py文件中解决它.

Answer 7

app = Celery('proj',
             broker='amqp://',
             backend='amqp://',
             include=['proj.tasks'])

请包括= ['proj.tasks'] 您需要转到顶部目录，然后执行此

celery -A app.celery_module.celeryapp worker --loglevel=info

不

celery -A celeryapp worker --loglevel=info

在您的celeryconfig.py中，输入imports =（“ path.ptah.tasks”，）

请在其他模块中调用任务！

Answer 8

使用--settings对我不起作用.我不得不使用以下内容来完成所有工作:

celery --config=celeryconfig --loglevel=INFO

这是添加了CELERY_IMPORTS的celeryconfig文件:

# Celery configuration file
BROKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'amqp://'

CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'America/Los_Angeles'
CELERY_ENABLE_UTC = True

CELERY_IMPORTS = ("tasks",)

我的设置有点棘手,因为我正在使用主管将celery作为守护进程启动.

Answer 9

对我来说,这个错误是通过确保包含任务的应用程序包含在django的INSTALLED_APPS设置下来解决的.

Answer 10

就我而言，问题是我的项目没有autodiscover_tasks正常进行。

在celery.py文件中，用于获取的代码是autodiscover_tasks：

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

我将其更改为以下内容：

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

祝你一切顺利。

Answer 11

当我向 django 应用程序添加一些信号处理时，我神秘地出现了这个问题。在此过程中，我将应用程序转换为使用 AppConfig，这意味着它不是简单地读取为'booking' in INSTALLED_APPS，而是读取为'booking.app.BookingConfig'.

Celery 不明白这意味着什么，所以我添加INSTALLED_APPS_WITH_APPCONFIGS = ('booking',)到我的 django 设置中，并修改了我celery.py的

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

到

app.autodiscover_tasks(
    lambda: settings.INSTALLED_APPS + settings.INSTALLED_APPS_WITH_APPCONFIGS
)

Answer 12

我在从 Celery Beat 运行任务时遇到了同样的问题。Celery 不喜欢相对导入，因此在 my 中celeryconfig.py，我必须显式设置完整的包名称：

app.conf.beat_schedule = {
   'add-every-30-seconds': {
        'task': 'full.path.to.add',
        'schedule': 30.0,
        'args': (16, 16)
    },
}