ViennaNET: набор библиотек для backend’а. Часть 2
Сообщество .NET-разработчиков Райффайзенбанка продолжает краткий разбор содержимого ViennaNET. О том, как и зачем мы к этому пришли, можно почитать в первой части.
В этой статье пройдемся по еще не рассмотренным библиотекам для работы с распределенными транзакциями, очередями и БД, которые можно найти в нашем репозитории на GitHub (исходники лежат здесь), а Nuget-пакеты здесь.
ViennaNET.Sagas
Когда в проекте происходит переход на DDD и микросервисную архитектуру, то при разнесении бизнес-логики по разным сервисам, возникает проблема, связанная с необходимостью реализации механизма распределенных транзакций, ведь многие сценарии часто затрагивают сразу несколько доменов. С такими механизмами подробнее можно познакомиться, например, в книге «Microservices Patterns», Chris Richardson.
В наших проектах мы реализовали простой, но полезный механизм: сага, а точнее сага на основе оркестрации. Суть ее в следующем: есть некий бизнес-сценарий, в котором необходимо последовательно совершить операции в разных сервисах, при этом, в случае возникновения каких-либо проблем на любом шаге, необходимо вызвать процедуру отката всех предыдущих шагов, где она предусмотрена. Таким образом, в конце выполнения саги, независимо от успешности, мы получаем консистентные данные во всех доменах.
Наша реализация пока сделана в базовом виде и не завязана на использовании каких-либо способов взаимодействия с другими сервисами. Применять её несложно: достаточно сделать наследника от базового абстрактного класса SagaBase<Т>, где T — это ваш класс контекста, в котором можно хранить исходные данные, необходимые для работы саги, а также некоторые промежуточные результаты. Экземпляр контекста будет пробрасываться во все шаги во время выполнения. Сама сага является stateless классом, поэтому экземпляр может быть помещен в DI как Singleton, чтобы получить необходимые зависимости.
Пример объявления:
public class ExampleSaga : SagaBase
{
public ExampleSaga()
{
Step("Step 1")
.WithAction(c => ...)
.WithCompensation(c => ...);
AsyncStep("Step 2")
.WithAction(async c => ...);
}
}
Пример вызова:
var saga = new ExampleSaga();
var context = new ExampleContext();
await saga.Execute(context);
Полноценные примеры разных реализаций можно посмотреть здесь и в сборке с тестами.
ViennaNET.Orm.*
Набор библиотек для работы с различными БД через Nhibernate. У нас используется подход DB-First с применением Liquibase, поэтому здесь присутствует только функционал по работе с данными в готовой БД.
ViennaNET.Orm.Seedwork и ViennaNET.Orm — главные сборки, содержащие базовые интерфейсы и их реализации соответственно. Остановимся на их содержимом подробнее.
Интерфейс IEntityFactoryService и его реализация EntityFactoryService являются главной отправной точкой для работы с БД, так как здесь создается Unit of Work, репозитории для работы с конкретными сущностями, а также исполнители команд и прямых SQL-запросов. Иногда удобно ограничить возможности класса по работе с БД, например, дать возможность только для чтения данных. Для таких случаев у IEntityFactoryService есть предок — интерфейс IEntityRepositoryFactory, в котором объявлен только метод для создания репозиториев.
Для непосредственного обращения к БД используется механизм провайдеров. Для каждой используемой у нас в командах СУБД есть своя реализация: ViennaNET.Orm.MSSQL, ViennaNET.Orm.Oracle, ViennaNET.Orm.SQLite, ViennaNET.Orm.PostgreSql.
При этом в одном приложении может быть зарегистрировано несколько провайдеров одновременно, что позволяет, например, в рамках одного сервиса без каких-либо затрат на доработку инфраструктуры провести пошаговую миграцию с одной СУБД на другую. Механизм выбора необходимого подключения и, следовательно, провайдера для конкретного класса-сущности (для которого и пишется маппинг на таблицы БД) реализован через регистрацию сущности в классе BoundedContext (содержит метод для регистрации доменных сущностей) или его наследника ApplicationContext (содержит методы для регистрации аппликационных сущностей, прямых запросов и команд), где в качестве аргумента принимается идентификатор подключения из конфигурации:
"db": [
{
"nick": "mssql_connection",
"dbServerType": "MSSQL",
"ConnectionString": "...",
"useCallContext": true
},
{
"nick": "oracle_connection",
"dbServerType": "Oracle",
"ConnectionString": "..."
}
],
Пример ApplicationContext:
internal sealed class DbContext : ApplicationContext
{
public DbContext()
{
AddEntity("mssql_connection");
AddEntity("oracle_connection");
AddEntity("oracle_connection");
}
}
Если идентификатор подключения не указан, то будет использоваться подключение с именем «default».
Непосредственно маппинг сущностей на таблицы БД реализуется стандартными средствами NHibernate. Можно использовать описание как через xml-файлы, так и через классы. Для удобного написания репозиториев-заглушек в Unit-тестах, имеется библиотека ViennaNET.TestUtils.Orm.
Полноценные примеры использования ViennaNET.Orm.* можно найти здесь.
ViennaNET.Messaging.*
Набор библиотек для работы с очередями.
Для работы с очередями был выбран такой же подход, что и с различными СУБД, а именно — максимально возможный унифицированный подход с точки зрения работы с библиотекой, независимо от используемого менеджера очередей. Библиотека ViennaNET.Messaging как раз отвечает за эту унификацию, а ViennaNET.Messaging.MQSeriesQueue, ViennaNET.Messaging.RabbitMQQueue и ViennaNET.Messaging.KafkaQueue содержат реализации адаптеров для IBM MQ, RabbitMQ и Kafka соответственно.
В работе с очередями есть два процесса: получение сообщения и отправка.
Рассмотрим получение. Здесь есть 2 варианта: для постоянного прослушивания и для получения единичного сообщения. Для постоянного прослушивания очереди необходимо для начала описать класс процессора, унаследованный от IMessageProcessor, который будет отвечать за обработку входящего сообщения. Далее его необходмо «привязать» к определенной очереди, делается это через регистрацию в IQueueReactorFactory с указанием идентификатора очереди из конфигурации:
"messaging": {
"ApplicationName": "MyApplication"
},
"rabbitmq": {
"queues": [
{
"id": "myQueue",
"queuename": "lalala",
...
}
]
},
Пример запуска прослушивания:
_queueReactorFactory.Register("myQueue");
var queueReactor = queueReactorFactory.CreateQueueReactor("myQueue");
queueReactor.StartProcessing();
Затем, при старте сервиса и вызова метода для начала прослушивания, все сообщения из указанной очереди будут попадать в соответствующий процессор.
Для получения единичного сообщения в интерфейсе-фабрике IMessagingComponentFactory есть метод CreateMessageReceiver, который создаст получателя, ожидающего сообщения из указанной ему очереди:
using (var receiver = _messagingComponentFactory.CreateMessageReceiver("myQueue"))
{
var message = receiver.Receive();
}
Для отправки сообщения необходимо воспользоваться всё той же IMessagingComponentFactory и создать отправителя сообщения:
using (var sender = _messagingComponentFactory.CreateMessageSender("myQueue"))
{
sender.SendMessage(new MyMessage { Value = ...});
}
Для сериализации и десереализации сообщения есть три готовых варианта: просто текст, XML и JSON, но при необходимости спокойно можно сделать свои реализации интерфейсов IMessageSerializer и IMessageDeserializer.
Мы постарались сохранить уникальные возможности каждого менеджера очередей, например, ViennaNET.Messaging.MQSeriesQueue позволяет отправлять не только текстовые, но и байтовые сообщения, а ViennaNET.Messaging.RabbitMQQueue поддерживает роутинг и создание очередей «на лету». В нашей обертке адаптера для RabbitMQ также реализовано некоторое подобие RPC: отправляем сообщение и ожидаем ответа из специальной временной очереди, которая создается только для одного ответного сообщения.
Вот пример использования очередей с основными нюансами подключения.
ViennaNET.CallContext
Мы используем очереди не только для интеграции между разными системами, но и для общения между микросервисами одного приложения, например, в рамках саги. Это привело к необходимости передачи вместе с сообщением таких вспомогательных данных, как логин пользователя, идентификатор запроса для сквозного логирования, ip-адрес источника и авторизационые данные. Для реализации пробрасывания этих данных разработали библиотеку ViennaNET.CallContext, которая позволяет хранить данные из входящего в сервис запроса. При этом то, каким образом был сделан запрос, через очередь или через Http, не играет роли. Затем, перед отправкой исходящего запроса или сообщения, достаются данные из контекста и помещаются в заголовки. Таким образом, следующий сервис получает вспомогательные данные и аналогично ими распоряжается.
Спасибо за внимание, ждём ваших комментариев и pull request-ов!
