如何使用 TTL 设置 Celery，以便从队列中清除旧消息？答案

【问题标题】：How to set Celery with TTL so it purges old messages from a queue?如何使用 TTL 设置 Celery，以便从队列中清除旧消息？
【发布时间】：2014-01-04 10:21:10
【问题描述】：

我正在使用 Celery、Django 和 RabbitMQ。我正在寻找一种方法来自动从 RabbitMQ 中的 celery 队列中清除旧消息。因此，每当默认创建队列时，它应该清除所有早于 x 秒的消息。

这些是我在 settings.py 中设置的变量。

CELERY_DEFAULT_EXCHANGE = 'celery'
CELERY_DEFAULT_EXCHANGE_TYPE = 'topic'
CELERY_DEFAULT_ROUTING_KEY = 'celery'
CELERY_DEFAULT_QUEUE = 'celery'
CELERY_EVENT_QUEUE_TTL= 30

我也试过这个，但它导致根本没有消息发送到 RabbitMQ。

CELERY_QUEUES = (                                                                                                                                                                 
Queue('celery', routing_key='celery', queue_arguments={'x-message-ttl': 30}),                                                                                                 
Queue('default', routing_key='default', queue_arguments={'x-message-ttl': 30}),                                                                                               
)

欢迎任何想法。

【问题讨论】：

标签： django rabbitmq celery django-celery

【解决方案1】：

x-message-ttl 以毫秒 - 不是秒为单位。

你想把这样的东西放在你的 celery 配置文件中：

from kombu import Exchange, Queue

CELERY_QUEUES = [
                     Queue(
                         'celery',
                         Exchange('celery'),
                         routing_key = 'celery',
                         queue_arguments = {
                                                'x-message-ttl': 30000
                                           }
                     )
                ]

【讨论】：

那么，celery 的默认队列配置就没用了吗？我不确定它们的用途是什么

【解决方案2】：

选项 CELERY_EVENT_QUEUE_TTL= 30 仅适用于事件，而不适用于名称所述的任务。只要有必要，Celery 就会为某项任务保留消息，这是有道理的。

在我的例子中，我使用了 RabbitMQ Federation，它使队列加倍。切换到 Shovel 为我解决了这个问题。

【讨论】：