работник не потребляет задачи после того, как сельдерей вызывается add_consumer
Я бы хотел использовать Celery (с RabbitMQ как backend MQ) для выполнения задач различного вкуса через разные очереди. Одно из требований заключается в том, что потребление (рабочими) из конкретной очереди должно иметь возможность приостанавливаться и возобновляться.
Сельдерей, похоже, обладает этой возможностью, вызывая add_consumer и cancel_consumer . Хотя мне удалось отменить потребление задач из очереди для конкретного работника, я не могу заставить работника возобновить потребление, вызвав add_consumer . Здесь приведен код для воспроизведения этой проблемы . Мое предположение, вероятно, я пропускаю какой-то параметр, который должен быть предоставлен либо в celeryconfig либо через аргументы при запуске рабочих?
- Прервать выполняемую задачу в сельдерей в django
- Соединение AMQP сброшено одноранговым узлом, но сельдерей подключен
- Почему я не могу установить соединение с rabbitMQ с помощью python?
- Зачем использовать сельдерей вместо RabbitMQ?
- Потребление очереди rabbitmq изнутри потоков python
Было бы здорово получить некоторые свежие пары глаз на этом. В Stackoverflow не так много обсуждений относительно add_consumer и Github. Поэтому я надеюсь, что здесь есть некоторые эксперты, желающие поделиться своими мыслями / опытом.
–
Я запускаю ниже:
ОС Windows, RabbitMQ 3.5.6, Erlang 18.1, Python 3.3.5, сельдерей 3.1.15
- Получение количества сообщений в очереди RabbitMQ
- Проекты Multi Celery с одним и тем же бэкэндовым процессом BrobMMM
- celery .delay зависает (последнее, а не проблема с auth)
- Использование сельдерея в качестве отказоустойчивого планировщика
- Очередь блокировки для заблокированных сообщений Selery / RabbitMQ?
One Solution collect form web for “работник не потребляет задачи после того, как сельдерей вызывается add_consumer”
Чтобы возобновить работу из очереди, вам нужно указать имя очереди, а также целевых работников. Вот как это сделать.
app.control.add_consumer(queue='high', destination=['celery@asus'])
Вот подпись add_consumer
def add_consumer(state, queue, exchange=None, exchange_type=None, routing_key=None, **options):
В вашем случае вы звоните с
app.control.add_consumer('high', destination=['celery@high1woka'])
Таким образом, high передается в состояние, а очередь пуста. Таким образом, он не может возобновить работу.