Временная очередь из сельдерея

Question

Я использую Сельдерей с RabbitMQ. В последнее время я заметил, что создается большое количество временных очередей.

Итак, я экспериментировал и обнаружил, что при сбое задачи (то есть при возникновении исключения) возникает временная очередь со случайным именем (например, c76861943b0a4f3aaa6a99a6db06952c) и очередь остается.

Некоторые свойства временной очереди, найденные в rabbitmqadmin, следующие:

auto_delete: True потребители: 0 прочный: ложный сообщений: 1 messages_ready: 1

И одна такая временная очередь создается каждый раз, когда задача не выполняется (то есть возникает исключение). Как избежать этой ситуации? Потому что в моей производственной среде формируется большое количество таких очередей.

Answer 1

Похоже, вы используете amqp в качестве бэкэнда результатов. Из документов есть подводные камни использования этой конкретной установки:

• Каждая новая задача создает новую очередь на сервере с тысячами задачи, которые брокер может быть перегружен очередями, и это повлияет на
  производительность в негативных отношениях. Если вы используете RabbitMQ, то каждый
  очередь будет отдельным процессом Erlang, поэтому, если вы планируете
  сохранить много результатов одновременно, возможно, вам придется увеличить Erlang
  предел процесса и максимальное количество файловых дескрипторов вашей ОС
  позволяет
• Старые результаты не будут очищаться автоматически, поэтому вы должны сделать обязательно использовать результаты, иначе количество очередей будет в конце концов выйти из-под контроля. Если вы используете RabbitMQ 2.1.1 или выше вы можете воспользоваться аргументом x-expires для очередей, который истечет очереди через определенный срок после того, как они неиспользованными. Срок действия очереди может быть установлен (в секундах) Настройка CELERY_AMQP_TASK_RESULT_EXPIRES (по умолчанию не включена).

Из того, что я прочитал в changelog , это больше не является бэкэндом по умолчанию в версиях> = 2.3.0, потому что пользователи получали немного в тылу этим поведением Я бы предложил изменить бэкэнд результатов, если это не та функциональность, которая вам нужна.

Answer 2

Ну, Филипп прямо здесь. Ниже приведено описание того, как я решил это. Это конфигурация в файле celeryconfig.py.

Я все еще использую CELERY_BACKEND = "amqp", как сказал Филипп. Но в дополнение к этому я сейчас использую CELERY_IGNORE_RESULT = True. Эта конфигурация гарантирует, что дополнительные очереди не будут сформированы для каждой задачи.

Я уже использовал эту конфигурацию, но все же, когда задача не выполняется, была сформирована дополнительная очередь. Затем я заметил, что использовал другую конфигурацию, которую нужно было удалить, которая была CELERY_STORE_ERRORS_EVEN_IF_IGNORED = True. Это привело к тому, что результаты не были сохранены для всех задач, а только для ошибок (задачи, которые не были выполнены) и, следовательно, одна дополнительная очередь для задачи, которая не выполнена.

Answer 3

CELERY_TASK_RESULT_EXPIRES определяет время жизни временных очередей. По умолчанию это 1 день. Вы можете изменить это значение.

Answer 4

amqp backend создает новую очередь для каждой задачи.Если вы хотите избежать этого, вы можете использовать rpc backend, который хранит результаты в одной очереди.

В вашей конфигурации установите

CELERY_RESULT_BACKEND = 'rpc'
CELERY_RESULT_PERSISTENT = True

. Подробнее о это на сельдерее документы .

Answer 5

Причина, по которой это происходит, в том, что удаленное управление рабочими сельдерея включено (по умолчанию включено).

Вы можете отключить его, установив для CELERY_ENABLE_REMOTE_CONTROL значение False Однако учтите, что вы потеряете возможность делать такие вещи, как add_consumer, cancel_consumer и т. Д., Используя celery command