带有rabbitmq的芹菜创建结果多个队列答案

【问题标题】：Celery with rabbitmq creates results multiple queues带有rabbitmq的芹菜创建结果多个队列
【发布时间】：2014-01-26 17:32:44
【问题描述】：

我已经用 RabbitMQ 安装了 Celery。问题是对于返回的每个结果，Celery 都会在 Rabbit 中创建，并在交换 celeryresults 中使用任务 ID 排队。

我仍然想要结果，但在 ONE 队列中。

我的芹菜配置：

from datetime import timedelta
OKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'amqp'
#CELERY_IGNORE_RESULT = True
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json', 'application/json']
CELERY_TIMEZONE = 'Europe/Oslo'
CELERY_ENABLE_UTC = True

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    'every-minute': {
        'task': 'tasks.remote',
        'schedule': timedelta(seconds=30),
        'args': (),
    },
}

这可能吗？怎么样？

谢谢！

【问题讨论】：

很难在没有看到你的代码的情况下猜测... celery 配置以及你如何创建队列、发送任务等。

标签： rabbitmq celery django-celery

【解决方案1】：

amqp 后端为每个任务创建一个新队列。或者，有一个新的 rpc 后端将结果保存在一个队列中。

http://docs.celeryproject.org/en/master/whatsnew-3.1.html#new-rpc-result-backend

【讨论】：

我是否需要为 Rabbit 配置一些特殊的东西才能监听 RPC？
不，不需要配置 Rabbit。将CELERY_RESULT_BACKEND = 'rpc://' 添加到您的celery_config.py，您应该会没事的。

【解决方案2】：

没什么异常。

当我们使用 amqp 作为结果后端时，这就是 celery 的工作原理。它将为每个结果对应于工人消耗的每个任务创建一个新的临时队列。

如果对结果不感兴趣，可以试试 CELERY_IGNORE_RESULT = True 设置

如果您确实想存储结果，那么我建议您使用不同的结果后端，例如 Redis。

【讨论】：

这个队列的临时性如何？我刚刚检查了一下，我有超过 10,000 个队列，其中包含超过 25,000 条消息。
我认为默认情况下是 24 小时.....您可以通过设置 CELERY_TASK_RESULT_EXPIRES= 来尝试。告诉我它是如何工作的？
虽然有 10000 个队列和 25000 条消息，但它看起来像是一个大型应用程序。请考虑其他后端。我也先尝试了rabbitmq result backend，后来由于这些原因，转移到redis。

【解决方案3】：

您说您希望 Celery 将结果保留在一个队列中。现在，为了回答你的问题，让我问你一个：

您如何期望每个生产者检查它的相关结果而不读取队列中的每条消息以找到它需要/想要的消息？

本质上，您需要的是一个键值对数据库，以便查找是 O(1)。使用队列代理的唯一方法是为每个“对”创建一个队列。

我知道拥有许多 GUID 队列并不整洁或漂亮，但从概念上讲，这是在消息传递代理上执行此操作的唯一方法。

【讨论】：

【解决方案4】：

此解决方案不会将所有结果保存到 ONE 队列中，但它至少会在您完成后立即清理多余的队列。

如果您使用 Redis 作为后端，当您完成创建错误队列的结果时，运行 result.forget()。这将导致结果和结果队列都消失。这可以帮助您管理队列数量，并防止出现 OOM 问题。

【讨论】：