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

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

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

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

Ответ 1

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

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

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

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

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

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

Ответ 2

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

Ответ 3

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

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

Ответ 4

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

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

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

Ответ 5

я работал на простом примере рабочего пула, используя rabbitmq. Это должно дать некоторую иллюстрацию, как этого достичь. https://github.com/kennykarnama/rabbitmq-worker-pool