День 1517. #ЗаметкиНаПолях
Параллельная Публикация Уведомлений в MediatR
MediatR — популярная библиотека с простой реализацией паттерна посредник в .NET. С ростом популярности паттерна CQRS MediatR стала популярной библиотекой для реализации команд и запросов.

Однако она также поддерживает паттерн издатель-подписчик с использованием уведомлений. Вы можете опубликовать экземпляр 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.

Источник: https://www.milanjovanovic.tech/blog/how-to-publish-mediatr-notifications-in-parallel