.NET Core开发实战（第35课：MediatR：让领域事件处理更加优雅）--学习笔记
.NET Core开发实战（第35课：MediatR：让领域事件处理更加优雅）--学习笔记
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;
}
}
这里实现里在执行命令之前判断事务是否开启，如果事务开启的话继续执行后面的逻辑，如果事务没有开启，先开启事务，再执行后面的逻辑