Django Celery delay() 总是推送到默认的“芹菜”队列

Posted

技术标签:

【中文标题】Django Celery delay() 总是推送到默认的“芹菜”队列【英文标题】:Django Celery delay() always pushing to default 'celery' queue 【发布时间】:2017-11-18 13:44:52 【问题描述】:

我正在用这个扯掉我的头发。

我的问题的症结在于,在我的 settings.py 中使用 Django CELERY_DEFAULT_QUEUE 设置不会强制我的任务进入我设置的特定队列。它总是进入我代理中的默认celery 队列。

但是,如果我在 shared_task 装饰器中指定 queue=proj:dev,它会进入正确的队列。它的行为符合预期。

我的设置如下:

我的本地主机上的 Django 代码(用于测试和其他东西)。通过 Django 的 shell (manage.py shell) 执行任务 .delay()'s 配置为我的代理的远程 Redis 实例 在远程机器上配置 2 个 celery worker 并等待来自 Redis 的消息(在 Google App Engine 上 - 可能不相关)

注意:对于下面的代码片段,我隐藏了项目名称并使用proj 作为占位符。

celery.py

from __future__ import absolute_import, unicode_literals

import os
from celery import Celery, shared_task

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

app = Celery('proj')

app.config_from_object('django.conf:settings', namespace='CELERY', force=True)

app.autodiscover_tasks()


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

settings.py

...
CELERY_RESULT_BACKEND = 'django-db'
CELERY_BROKER_URL = 'redis://:@:6379/0'.format(
    os.environ.get('REDIS_PASSWORD'),
    os.environ.get('REDIS_HOST', 'alice-redis-vm'))
CELERY_DEFAULT_QUEUE = os.environ.get('CELERY_DEFAULT_QUEUE', 'proj:dev')

我的想法是,就目前而言,我希望为我的代码所在的不同环境设置不同的队列:dev、staging、prod。因此,在 Google App Engine 上,我定义了一个基于单个 App Engine 服务传递的环境变量。

步骤

因此,通过上述配置,我使用./manage.py shell 启动shell 并运行add.delay(2, 2)。我收到了 AsyncResult 回复,但 Redis 监视器清楚地显示一条消息已发送到默认的 celery 队列:

1497566026.117419 [0 155.93.144.189:58887] "LPUSH" "celery"
...

我错过了什么?

不要在工作中扔扳手,但我觉得今天有一个点,这实际上是有效的。但就我的一生而言,我想不出我的大脑的哪一部分让我失望了。

堆栈版本:

python:3.5.2 celery:4.0.2 redis:2.10.5 django:1.10.4

【问题讨论】:

嗨 Kyle,您确定 CELERY_DEFAULT_QUEUE 是 redis 后端支持的设置吗?从文档中,它仅列在 AMQP 后端设置部分下。对于您指定的情况,您可能可以为每个环境指定不同的数据库。例如,0=dev,1=staging,2=production @e4c5 我会接受这个赌注。如果 env 缺少它,它将默认为 'proj:dev'。 os.environ.get 的第二个值如果不存在则为默认值。 Kyle 确保您在 CELERY_QUEUES 设置中定义了 proj:dev @e4c5 - 所以CELERY_DEFAULT_QUEUE 是一个常数。它将由一个也称为CELERY_DEFAULT_QUEU(一致性FTW)的环境变量设置。但是,Robin 是对的,它只会默认为 proj:dev。然而,尽管默认,我的消息不会发送到proj:dev @hurturk - 那是缺失的,我会添加一些配置,再试一次并在必要时更新问题,谢谢! 【参考方案1】:

这个问题比我想象的要简单得多——文档不正确!!

Celery 文档要求我们使用 CELERY_DEFAULT_QUEUE 在 celery 对象上设置 task_default_queue 配置。

参考:http://docs.celeryproject.org/en/latest/userguide/configuration.html#new-lowercase-settings

我们目前应该使用CELERY_TASK_DEFAULT_QUEUE。这是所有其他设置名称的命名不一致。它是在 Github 上提出的 - https://github.com/celery/celery/issues/3772

解决方案总结

在配置模块中使用CELERY_DEFAULT_QUEUE(使用config_from_object)对队列没有影响。

请改用CELERY_TASK_DEFAULT_QUEUE

【讨论】:

TIL celery 配置是所有变量的小写映射。谢谢。【参考方案2】:

如果您来到这里是因为您尝试在 Celery 中使用 SQS 实现一个预定义的队列,并且发现 Celery 在 SQS 中创建了一个名为“celery”的新队列,无论您说什么,您的旅程都到了终点朋友。

在将broker_transport_options 传递给 Celery 之前,更改您的默认队列和/或指定您将明确使用的队列。就我而言,我只需要一个队列,因此可以执行以下操作:

celery.conf.task_default_queue = "<YOUR_PREDEFINED_QUEUE_NAME_IN_SQS">

【讨论】:

以上是关于Django Celery delay() 总是推送到默认的“芹菜”队列的主要内容,如果未能解决你的问题,请参考以下文章

Celery 任务在 Django 测试中没有抛出异常

django celery get() 用于获取结果表单任务

使用 django、celery 和 redis 的一项一项任务

使用类方法作为 celery 任务

异步任务队列Celery在Django中的使用

celery delay 没反应