35 | MediatR：让领域事件处理更加优雅

核心对象

IMediator

INotification

INotificationHandler

这两个与之前的 Request 的行为是不一样的，接下来看一下代码

internal class MyEvent : INotification
{public string EventName { get; set; }
}internal class MyEventHandler : INotificationHandler<MyEvent>
{public Task Handle(MyEvent notification, CancellationToken cancellationToken){Console.WriteLine($"MyEventHandler执行：{notification.EventName}");return Task.CompletedTask;}
}internal class MyEventHandlerV2 : INotificationHandler<MyEvent>
{public Task Handle(MyEvent notification, CancellationToken cancellationToken){Console.WriteLine($"MyEventHandlerV2执行：{notification.EventName}");return Task.CompletedTask;}
}

//await mediator.Send(new MyCommand { CommandName = "cmd01" });
await mediator.Publish(new MyEvent { EventName = "event01" });

之前 mediator 使用了 Send 的方式来处理 Command，它还有一个方法 Publish，这个方法的入参是一个 INotification

启动程序，输出如下：

MyEventHandler执行：event01
MyEventHandlerV2执行：event01

与之前的 IRequest 不同的是，INotification 是可以注册多个 Handler 的，它是一个一对多的关系，借助它就可以对领域事件定义多个处理器来处理

接着看一下之前云服务的代码

public async Task<bool> SaveEntitiesAsync(CancellationToken cancellationToken = default)
{var result = await base.SaveChangesAsync(cancellationToken);await _mediator.DispatchDomainEventsAsync(this);return true;
}

之前在 IUnitOfWork 定义的时候讲过一个发送领域事件的方法 DispatchDomainEventsAsync，看一下这个方法的定义

static class MediatorExtension
{public static async Task DispatchDomainEventsAsync(this IMediator mediator, DbContext ctx){var domainEntities = ctx.ChangeTracker.Entries<Entity>().Where(x => x.Entity.DomainEvents != null && x.Entity.DomainEvents.Any());var domainEvents = domainEntities.SelectMany(x => x.Entity.DomainEvents).ToList();domainEntities.ToList().ForEach(entity => entity.Entity.ClearDomainEvents());foreach (var domainEvent in domainEvents)await mediator.Publish(domainEvent);}
}

可以看到这里是将所有的实体内的领域事件全部都查找出来，然后通过 mediator 的 Publish 发送领域事件，具体的领域事件的处理注册在 mediator 里面，这里定义了一个 OrderCreatedDomainEventHandler

public class OrderCreatedDomainEventHandler : IDomainEventHandler<OrderCreatedDomainEvent>
{ICapPublisher _capPublisher;public OrderCreatedDomainEventHandler(ICapPublisher capPublisher){_capPublisher = capPublisher;}public async Task Handle(OrderCreatedDomainEvent notification, CancellationToken cancellationToken){await _capPublisher.PublishAsync("OrderCreated", new OrderCreatedIntegrationEvent(notification.Order.Id));}
}

它继承自 IDomainEventHandler，而 IDomainEventHandler 继承自 INotificationHandler

public interface IDomainEventHandler<TDomainEvent> : INotificationHandler<TDomainEvent>where TDomainEvent : IDomainEvent
{//这里我们使用了INotificationHandler的Handle方法来作为处理方法的定义Task Handle(TDomainEvent domainEvent, CancellationToken cancellationToken);
}

这也就是为什么 IDomainEventHandler 会识别到 DomainEvent 并且进行处理，同样的在定义 DomainEvent 的时候，也需要标识它是一个 DomainEvent

public class OrderCreatedDomainEvent : IDomainEvent
{public Order Order { get; private set; }public OrderCreatedDomainEvent(Order order){this.Order = order;}
}

而 DomainEvent 实际上也是继承自 INotification

public interface IDomainEvent : INotification
{
}

这也就意味着 EventHandler 可以正确的识别到对应的 Event 并且进行处理，这都是 MediatR 的核心能力

领域事件都是定义在 event 目录下，与领域模型定义在一起，所有的领域事件都继承 DomainEvent，分布于这个目录

领域事件的处理 Handler 都定义在 Application 应用层的 Application 下面的 DomainEventHandlers 目录下面

这样的好处是事件的定义与事件的处理是分开的，并且非常的明确知道有哪些领域事件，有哪些领域事件的处理程序

关于 MediatR 再补充一部分内容，在 TransactionBehavior 内可以看到这个类实际上继承自 IPipelineBehavior

namespace MediatR
{public interface IPipelineBehavior<in TRequest, TResponse>{Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate<TResponse> next);}
}

这个接口的作用是在命令或者事件处理的之前或者之后插入逻辑，它的执行的方式有点像中间件的方式，在 Handler 的入参里面有一个 next 的参数，就是指 CommandHandler 或者 EventHandler 的执行的逻辑，在这里就可以决定 Handler 的具体执行之前或者之后，插入一些逻辑

public async Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate<TResponse> next)
{var response = default(TResponse);var typeName = request.GetGenericTypeName();try{// 首先判断当前是否有开启事务if (_dbContext.HasActiveTransaction){return await next();}// 定义了一个数据库操作执行的策略，比如说可以在里面嵌入一些重试的逻辑，这里创建了一个默认的策略var strategy = _dbContext.Database.CreateExecutionStrategy();await strategy.ExecuteAsync(async () =>{Guid transactionId;using (var transaction = await _dbContext.BeginTransactionAsync())using (_logger.BeginScope("TransactionContext:{TransactionId}", transaction.TransactionId)){_logger.LogInformation("----- 开始事务 {TransactionId} ({@Command})", transaction.TransactionId, typeName, request);response = await next();// next 实际上是指我们的后续操作，这里的模式有点像之前讲的中间件模式_logger.LogInformation("----- 提交事务 {TransactionId} {CommandName}", transaction.TransactionId, typeName);await _dbContext.CommitTransactionAsync(transaction);transactionId = transaction.TransactionId;}});return response;}catch (Exception ex){_logger.LogError(ex, "处理事务出错 {CommandName} ({@Command})", typeName, request);throw;}
}

这里实现里在执行命令之前判断事务是否开启，如果事务开启的话继续执行后面的逻辑，如果事务没有开启，先开启事务，再执行后面的逻辑