Как установить сельдерей с TTL, чтобы он удалял старые сообщения из очереди?

Question

Я использую сельдерей, Django и RabbitMQ. Я ищу способ очистить старые сообщения из очереди сельдерея в RabbitMQ автоматически. Поэтому всякий раз, когда очередь создается по умолчанию, она должна очищать все сообщения старше х секунд.

Это переменные, заданные в моих настройках.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}),                        
) 

Любая идея приветствуется.

Answer 1

Параметр CELERY_EVENT_QUEUE_TTL = 30 предназначен только для событий, а не для задач, как указано в названии. Сельдерей будет хранить сообщение для выполнения задачи, если это необходимо, что имеет смысл.

В моем случае я использовал федерацию RabbitMQ, которая удвоила очередь. Переключение на Лопату решило проблему для меня.

Answer 2

х-сообщение-Время_жизни в миллисекундах - НЕ СЕК.

Вы хотите положить что-то подобное в вашем сельдерея конфигурационном файле:

from kombu import Exchange, Queue 

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