Рабочие пулы и многопользовательские очереди с RabbitMQ

Question

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

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

Есть ли относительно прямой путь для достижения этой цели? Я довольно новичок в RabbitMQ и на самом деле не смог достичь того, что нам нужно. Мы также не хотим писать очень сложное многопоточное потребительское приложение, это время, затрачиваемое на разработку и тестирование, которое мы, вероятно, не можем себе позволить. Наш стек основан на Windows / .Net / C #, если это germaine, но я не думаю, что это должно иметь главное значение в данном вопросе.

Answer 1

Вы можете посмотреть на реализацию очереди приоритетов (которая не была реализована, когда этот вопрос был задан изначально): https://www.rabbitmq.com/priority.html

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

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

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

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

Первое предложение со 100 очередями также может иметь проблему, если для одного пользователя существует 100 медленных задач, а затем другой пользователь публикует пакет задач. Эти задачи не будут рассмотрены, пока одна из медленных задач не будет завершена. Если это окажется законной проблемой, вы потенциально можете объединить два решения.

Answer 2

Я не понимаю, почему вы не используете vhosts RabbitMQ, и ваше приложение входит в RabbitMQ и проходит аутентификацию по отдельному соединению для каждого пользователя.

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

Answer 3

Вы можете просто иметь свой пул рабочих, которые будут использовать одну и ту же уникальную очередь. Затем работа будет распределена между ними, и вы сможете увеличивать / уменьшать свой пул, чтобы увеличивать / уменьшать производительность вашей работы.

Answer 4

Рабочим назначается 0+ очередей, а не обменов.

Логика, для которой будут взяты задачи, из которых очереди для каждого работника реализованы в классе, указанном через CELERYD_CONSUMER, который по умолчанию celery.worker.consumer.Consumer.

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