Распределите сообщения rabbitmq равномерно

Question

На данный момент у нас есть ряд издателей (микросервисов), которые публикуют свои сообщения для обмена.Каждое сообщение имеет атрибут serviceId.Очередь подключена к одному абоненту (микросервис), который обрабатывает сообщения очереди, обработка одного сообщения является дорогостоящей операцией (занимает около 20-30 секунд).В настоящее время мы имеем следующую ситуацию: служба A публикует ~ 200 сообщений, через несколько секунд служба B публикует 2 сообщения.Таким образом, подписчик будет обрабатывать эти 2 сообщения только после того, как будут обработаны первые 200.

Мы хотим обрабатывать сообщения в порядке их поступления в очередь, но с учетом идентификатора источника службы.

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

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

Каков наилучший подход для решения этой ситуации?Спасибо!

Answer 1

/! \ Будьте осторожны, издатели публикуют в обмен, а не в очередь.

Мы хотим обрабатывать сообщения в порядке их поступления в очередь, но относительно источникаserviceId.

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

Решение, которое я бы предложил здесь, - это прямой обмен ключами маршрутизации, такими как xxxxx.<serviceId>.Затем вы можете привязать одну очередь по serviceId (то есть: одну очередь для службы A, одну для службы B, ...), причем каждый потребитель будет использовать все очереди.

Затем необходимо обработать подписку издателя.Я бы заставил издателя опубликовать сообщение "привет", причем это сообщение потребляется каждым потребителем, который, в свою очередь, связывает новую очередь для этой службы (используя xxxxx.<newServiceId>), и, наконец, опубликовать ответ обратно (чтобы издатель могначать отправку сообщений).

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

Надеюсь, это поможет.