⚡️ Параллельная Публикация Уведомлений в MediatR
MediatR — это популярная библиотека с простой реализацией паттерна посредник в .NET.Посредник — паттерн шаблон проектирования, обеспечивающий взаимодействие множества объектов, формируя при этом слабое зацепление и избавляя объекты от необходимости явно ссылаться друг на друга.
С ростом популярности паттерна CQRS MediatR стала популярной библиотекой для реализации команд и запросов.
Издатель — подписчик — поведенческий шаблон проектирования передачи сообщений, в котором отправители сообщений, именуемые издателями (англ. publishers), напрямую не привязаны программным кодом отправки сообщений к подписчикам (англ. subscribers).
Но она также поддерживает паттерн издатель-подписчик с использованием уведомлений. Можно опубликовать экземпляр INotification, и несколько подписчиков обработают опубликованное сообщение.
Разберемся как это можно делать параллельно.
Нам нужен класс, реализующий INotification:
public record OrderCreated(Guid OrderId) : INotification;
А также реализация соответствующего INotificationHandler:public class OrderCreatedHandler :
INotificationHandler<OrderCreated>
{
private readonly INotificationService svc;
public OrderCreatedHandler(
INotificationService service)
{
svc = service;
}
public async Task Handle(
OrderCreated notification,
CancellationToken ct)
{
await svc.SendOrderCreatedEmail(
notification.OrderId,
ct);
}
}
Теперь можно публиковать сообщение с помощью IMediator или IPublisher:
await publisher.Publish(
new OrderCreated(order.Id),
cancellationToken);
MediatR вызовет все соответствующие обработчики. До 12й версии MediatR стратегия публикации вызывала каждый обработчик по отдельности. Однако появился новый интерфейс INotificationPublisher, управляющий тем, как вызываются обработчики.
Реализация по умолчанию - ForeachAwaitPublisher:
public class ForeachAwaitPublisher
: INotificationPublisher
{
public async Task Publish(
IEnumerable<NotificationHandlerExecutor> executors,
INotification notification,
CancellationToken ct)
{
foreach (var e in executors)
{
await e
.HandlerCallback(notification, ct)
.ConfigureAwait(false);
}
}
}
Она вызывает обработчики по одному и завершается неудачей при ошибке в одном из обработчиков.
Но вы также можете использовать TaskWhenAllPublisher (показаны только отличия в реализации метода Publish:
var tasks = executors
.Select(e =>
e.HandlerCallback(notification, ct))
.ToArray();
return Task.WhenAll(tasks);
TaskWhenAllPublisher вызывает все обработчики одновременно и выполняет их все независимо от того, возникали ли в них ошибки. Если вы сохраните задачу, возвращенную TaskWhenAllPublisher, вы можете получить доступ к свойству Task.Exception, содержащему экземпляр AggregateException, и реализовать обработку исключений.
Настройка стратегии публикации происходит в методе AddMediatR.
Если вы хотите использовать стратегию TaskWhenAllPublisher, вы можете:
- указать значение для свойства NotificationPublisher (тогда издатель будет синглтоном),
- указать тип стратегии в свойстве NotificationPublisherType и использовать свойство ServiceLifetime для задания времени жизни:
services.AddMediatR(cfg => {
…
cfg.NotificationPublisher =
new TaskWhenAllPublisher();
// или
cfg.NotificationPublisherType =
typeof(TaskWhenAllPublisher);
cfg.ServiceLifetime = ServiceLifetime.Transient;
});
Вы также можете реализовать пользовательский экземпляр INotificationPublisher и вместо этого использовать собственную реализацию.
Плюсы
Параллельного запуска обработчиков уведомлений обеспечивает значительное повышение производительности по сравнению с поведением по умолчанию.
Однако обратите внимание, что все обработчики будут использовать одну и ту же область видимости. Если у вас есть экземпляры сервисов, которые не поддерживают конкурентный доступ, вы можете столкнуться с проблемами. К сожалению, одним из таких является EF Core DbContext.
📌 Подробнее
@csharp_1001_notes
>>Click here to continue<<