Celeryev Очередь в RabbitMQ становится очень большой

Я использую сельдерей на кроличнике. Я отправлял тысячи сообщений в очередь, и они обрабатываются успешно, и все работает нормально. Однако количество сообщений в нескольких очередях rabbitmq растет довольно много (сотни тысяч единиц в очереди). Очереди называются celeryev.[...] (см. Снимок экрана ниже). Это подходящее поведение? Какова цель этих очередей, и не следует ли их регулярно чистить? Есть ли способ чистить их более регулярно, я думаю, что они занимают довольно много дискового пространства.

screenshot of rabbitmq control

Ответы

Ответ 1

Вы можете использовать CELERY_EVENT_QUEUE_TTL вариант сельдерея (только для работы с amqp), который установит время истечения сообщения, после чего оно будет удалено из очереди.

Ответ 2

Для всех, кто сталкивается с проблемами с очередью celeryev, становясь очень большой и угрожая дисковым пространством на вашем сервере rabbitmq, остерегайтесь принятого ответа! Вот мое предложение. Просто выполните эту команду в своем экземпляре rabbitmq:

rabbitmqctl set_policy limit_celeryev_queues "^celeryev\." '{"max-length":1000000}' --apply-to queues

Это ограничит любую очередь, начиная с "ceralev" до 1 миллиона записей. Я экспериментировал с застрявшим экземпляром flower, вызывающим очередь сбегающей зернистой решетки, и установил CELERY_EVENT_QUEUE_TTL/CELERY_EVENT_QUEUE_EXPIRES помог не управлять размером очереди.

В моем тестировании я начал процесс flower, затем SIGSTOP'ил его и наблюдал, как очередь celeryevev начинает убегать. Ни одна из этих настроек не помогла. Я подтвердил, что SIGCONT в процессе flower быстро вернет очередь в 0. Я не уверен, почему эти две кнопки не помогли, но это может иметь какое-то отношение к тому, как RabbitMQ реализует эти два параметра.

Во-первых, Per-Message TTL, соответствующая CELERY_EVENT_QUEUE_TTL, устанавливает только время истечения срока действия для каждой записи в очереди - AIUI не будет автоматически удалите сообщение из очереди, чтобы сэкономить место по истечении срока действия. Во-вторых, Queue TTL, соответствующий CELERY_EVENT_QUEUE_EXPIRES, говорит, что он "... гарантирует, что очередь будет удалена, если она не используется, по крайней мере, для срок действия". Тем не менее, я считаю, что их определение "неиспользуемых" может быть слишком строгим, чтобы использовать его, например. перегруженный, застрявший или убитый процесс цветка.

EDIT: К сожалению, одна проблема с этим предложением состоит в том, что set_policy ... apply-to queues будет влиять только на существующие очереди, а цветок может и будет создавать новые очереди, которые могут переполняться.

Ответ 3

Использование сельдерея celeryev префиксные очереди (и обмен) для monitoring, вы можете настроить его так, как хотите или отключить вообще ( celery control disable_events).

Ответ 4

Вам просто нужно установить конфигурацию для вашего Сельдерея.

Если вы хотите, чтобы сельдерей не создавал очереди celeryev.*:

CELERY_SEND_EVENTS = False # Will not create celeryev.* queues

Если вам нужны эти очереди для целей мониторинга (например, CeleryFlower), вы можете регулярно их чистить:

CELERY_EVENT_QUEUE_EXPIRES = 60 # Will delete all celeryev. queues without consumers after 1 minute.

Решение получилось здесь: https://www.cloudamqp.com/docs/celery.html