Создание очередей динамически с Celery

Question

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

Одиночная очередь вызывает проблему справедливости: если сообщение входит в большой список рассылки, большое количество задач добавляется в очередь, и другие сообщения не могут попасть в другие, более мелкие списки рассылки, пока все сообщения не большой список был отправлен. Как я могу найти способ обойти это?

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

Answer 1

Как показано на рисунке ниже, рассмотрим систему, которая использует обмен тем вместо прямого обмена.

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

Систему можно настроить так, чтобы высокоприоритетные сообщения передавались в выделенную очередь и потребителю, аналогично обычные сообщения или сообщения с низким приоритетом обрабатываются одной или несколькими очередями.

Celery поддерживает балансировку нагрузки для нескольких рабочих (количество рабочих зависит от оборудования). Настройки конфигурации сельдерея, такие как BROKER_POOL_LIMIT, PREFETCH_LIMIT и т. Д., Помогают улучшить балансировку нагрузки и уменьшить перегрузку.