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;
}
}
這里實現里在執行命令之前判斷事務是否開啟，如果事務開啟的話繼續執行后面的邏輯，如果事務沒有開啟，先開啟事務，再執行后面的邏輯
GitHub源碼鏈接：
https://github.com/witskeeper/geektime
本作品采用知識共享署名-非商業性使用-相同方式共享 4.0 國際許可協議進行許可。
歡迎轉載、使用、重新發布，但務必保留文章署名 鄭子銘 （包含鏈接： http://www.cnblogs.com/MingsonZheng/ ），不得用於商業目的，基於本文修改后的作品務必以相同的許可發布。
如有任何疑問，請與我聯系 (MingsonZheng@outlook.com) 。