Как настроить MassTransit + RabbitMQ, чтобы несколько потоков потребления получали все копии опубликованных сообщений?

Question

[Руслан]

Здравствуйте!

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

я настраиваю очередь так:

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                var host = cfg.Host(new Uri("rabbitmq://localhost/"), h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                });

                cfg.ReceiveEndpoint(host,"customer_update_queue", ec =>
                {
                    ec.LoadFrom(_instance);
                });
            });

            _instance.RegisterInstance<IBusControl>(busControl);
            _instance.RegisterInstance<IBus>(busControl);

            busControl.Start();

_instance - это IoC контейнер Unity

в других процессах создание и подключение к шине выглядит так:

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                var host = cfg.Host(new Uri("rabbitmq://localhost/"), h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                });

                cfg.ReceiveEndpoint(host, "customer_update_queue", ec =>
                {
                    //ec.Consumer<SmsService>();
                    SmsService s = new SmsService();
                    ec.Instance(s);
                });
            });


            busControl.Start();

Спасибо за внимание.

Answer 1

[Иван Шумов]

Перестаньте настраивать архитектуру кроля в коде. Это к хорошему не приведет. Зайдите к ним на сайт - там отличные туториалы. А вообще - вам нужен exchange к котором присоединено несколько очередей

Comments

[Руслан]

Правильно ли я понимаю, что вы рекомендуете отдельно настроить структуру обменников (exchange) и очередей, так, чтобы сообщения между ними правильно передавались, а потом без лишнего конфигурирования привязать приложения к (чему?) обменникам и/или очередям?

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

[Иван Шумов]

Руслан, если у вас очереди временные то тогда можете создавать их на стороне клиента и прикреплять их к существующему exchange. Учтите только что временные очереди существуют только пока клиент жив. Для всего остального используйте фиксированную конфигурацию. Есть некоторые дополнительные юзкейсы, но и по вашим словам не очень понятно для чего вы используете RMQ.

И выражайтесь, пожалуйста, терминами RMQ - exchange, queue, ...

[Руслан]

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

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

[Иван Шумов]

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

[Руслан]

На сайте MassTransit не очень много информации относительно конфигурирования RMQ.
Есть вот такая статья: masstransit-project.com/MassTransit/understand/def...
там, вроде бы для получения exchange не нужен, но, как мне кажется, поскольку я использую MassTransit, он сам создает exchange для каждой очереди, т.к. у него транспорт двунаправленный.

[Иван Шумов]

Руслан, а вам оно вообще надо как прослойка с RMQ? тыц

[Руслан]

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

[Иван Шумов]

Руслан, то есть вместо одного из лучших решений на рынке вы выбрали неподдерживаемый самопал? Умно-умно

[Руслан]

"вместо одного из лучших решений" - это вы имеете в виду RMQ?

[Иван Шумов]

Руслан, на рынке сейчас 2 топовых решения по Message Broker - RabbitMQ и Kafka. Все остальное пока что даже начинать смотреть не стоит

[Руслан]

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

[Иван Шумов]

Руслан, только это не возможно. Одни брокеры push based, другие pull based

[Руслан]

Не понимаю, что вы имеете в виду. Тестовые примеры у меня вроде бы отработали как надо.
Спасибо вам за подсказки.