Как работать с MassTransit и RabbitMQ: реализуем RPC с Saga и без

Привет, Хабр! Меня зовут Евгений, я backend-разработчик SimbirSoft. В этой статье я разберу два варианта решения нетривиальной задачи создания RPC через брокер сообщений RabbitMQ и библиотеку MassTransit. Подробно разберём подключение MassTransit и работу с Saga. Тема будет полезна как для начинающих, так и опытных backend-разработчиков .NET.

Введение

В микросервисной архитектуре общение между отдельными сервисами как правило делится на два типа:

1. Синхронное (Request-Response) применяется, когда на запрос нужен незамедлительный ответ (таймаут соединения в расчёт не берётся).

2. Асинхронное (

   Мы рассмотрим работу брокера RabbitMQ и асинхронную модель взаимодействия сервисов. Подробнее о его особенностях в сравнении с его конкурентом Apache Kafka можно почитать здесь.

   Итак, официальная документация RabbitMQ гласит, что брокер сообщений — это сервер промежуточного ПО между автором (Producer)  и получателем сообщений (Consumer), который выполняет функции получения сообщения от автора, маршрутизации и установки сообщений в очередь получателя сообщения.

   Для наглядности ниже приведена принципиальная схема устройства брокера сообщений.

   

   При использовании AMQP асинхронность достигается за счет очередей сообщений. То есть если получатель сообщений (Consumer) занят, то ему не нужно всё бросать и хвататься обрабатывать сообщение, он сможет обработать сообщение из очереди, когда будет к этому готов.

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

   

   Применение брокера сообщений упорядочивает и упрощает взаимодействие сервисов между собой.

   А вот представление схемы взаимодействия с картинки выше, если бы использовался брокер сообщений:

   

   Существует две основные архитектуры брокеров сообщений:

   1. Push-архитектура. Если у сервера (брокера) есть сообщения для получателя (Consumer), то он по своей инициативе отправляет сообщение получателю.

   

   2. Pull-архитектура. Клиент (он же Consumer) по своей инициативе запрашивает у сервера (брокера) сообщения, и если у брокера есть сообщения для клиента, то отправляет эти сообщения.

   

   В данной статье будет рассмотрена Push-архитектура, наиболее распространенная при использовании брокера сообщений RabbitMQ.

   От слов к делу!

   Реализуем RPC c помощью MassTransit

   Наиболее популярные библиотеки для работы с RabbitMQ в .NET:

   • RabbitMQ.Client

   • MassTransit.RabbitMQ

   • EasyNetQ

   • NServiceBus.RabbitMQ

   В качестве примера работы с RabbitMQ рассмотрим нестандартную задачу построения RPC (Remote Procedure Call, удаленный вызов процедур) c использованием библиотеки MassTransit.

   Условия задачи:

   Программа генерирует число, которое пользователь должен угадать. При каждом вводе числа программа пишет результат: больше или меньше отгадываемого. Количество попыток угадывания и диапазон чисел должны задаваться из настроек. 

   Рассмотрим два варианта решения:

   • без использования Saga

   • с использованием Saga

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

   Краткое описание реализованной функциональности:

   • Консольный клиент принимает от пользователя число и отправляет его в брокер сообщений.

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

   • Клиент получает от брокера сообщений ответ сервера и выводит его пользователю.

   Полный код проекта приведен на github

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

   Решение без использования Saga MassTransit

   Принципиальная схема приложения без использования Saga:

   

   Теперь переходим в Visual Studio и создаем наши проекты.

   

   Настройки подключения к облаку RabbitMQ указаны в appsettings.json файлах:

   

   При старте Producer-приложения и Consumer-приложения происходит инициализация настроек для подключения к RabbitMQ.

   Для Producer:

   

   И для Consumer:

   

   

   Для примера работы с брокером сообщений мы используем библиотеку MassTransit.RabbitMQ v. 8.1.3

   В данном проекте вся логика работы с брокером представлена в библиотеке классов «Infrastructure.MassTransit»:

   

   Процесс отправки сообщения из консольного клиента:

   

   Процесс получения сообщений из очереди ответов для консольного клиента:

   

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

   

   Для прослушивания сообщений из очереди ответов для консольного клиента определен класс RabbitMqConsumer, который реализует интерфейс IConsumer библиотеки MassTransit.

   Логика работы нашего консольного клиента лишь в отправке числа, которое ввел пользователь через консоль на сервер и получение ответа (более подробно можно посмотреть в github), поэтому класс RabbitMqConsumer лишь получает сообщения и выставляет флаг, отгадано ли число с сервера или нет:

   

   Теперь рассмотрим request от клиента к серверу через брокер и реализацию responce от сервера к клиенту через брокер:

   В классе Program проекта WebApi.Consumer указан метод расширения, внутри которого, помимо установки настроек подключения к облаку RabbitMQ, также указываются настройки класса, который будет прослушивать сообщения из очереди для сервера:

   

   Метод AddBLLServicesWithoutSaga():

   

   Настройка подключения к облаку RabbitMQ и настройка класса прослушивания входящих сообщений:

   

   Так как мы реализуем RPC, то класс, который получает сообщения из очереди от клиента, должен обрабатывать данные и возвращать ответ клиенту через очередь ответов. Это всё реализовано в классе RabbitMqConsumerWebApi:

   

   Этот класс реализует интерфейс IConsumer из библиотеки MassTransit.

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

   Более подробное описание работы программы указано в файле Readme на github.

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

   

   Далее откроется страница управления RabbitMQ c довольно богатой функциональностью, если мы перейдем во вкладку очередей, то увидим наши очереди:

   

   Решение с использованием Saga от MassTransit

   Библиотека MassTransit имеет возможность организовать и управлять серий событий.

   Схема приложения с использованием Saga от MassTransit:

   

   Для использования Saga изменим класс Program консольного клиента:

   

   Внутри метода расширения устанавливаем настройки типов запроса и настройки конфигурации:

   

   Для отправки сообщений и получения ответа от саги необходимо использовать метод GetResponse интерфейса IRequestClient библиотеки MassTransit:

   

   

   Далее создаем проект ASP.NET Core c названием WebApi.Saga, внутри которого реализуем сагу для управления запросами и ответами наших микросервисов. В классе Program проекта ASP.NET Core добавим настройки конфигурации:

   

   

   Далее необходимо создать класс, описывающий состояние саги:

   

   И класс, описывающий обработку событий, возникающих в саге:

   

   Теперь отредактируем класс-обработчик сообщений от клиента:

   

   Также стоит отметить, что при запуске программы в облаке RabbitMQ  автоматически создадутся очереди, которыми будет управлять сага:

   

   Вывод

   При взаимодействии ряда микросервисов между собой предпочтительным вариантом является использование Saga от MassTransit в силу того, что Saga представляет мощную функциональность управления распределенными транзакциями. Помимо этого имеется возможность создавать и обрабатывать события. Более подробно с Saga можно ознакомиться здесь.

   Надеюсь, теперь вы получили представления о работе брокера сообщений RabbitMQ по Push-архитектуре в связке с библиотекой Masstransit. А реализовать это на практике помогут два шаблона проекта реализации RPC (с использованием Saga и без).

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

   Больше авторских материалов для backend-разработчиков от моих коллег читайте в соцсетях SimbirSoft – ВКонтакте и Telegram.