概述
利用發布訂閱模式來解耦不同架構層級，亦可用于解決隔離業務之間的互動
優點：
- 松耦合
- 橫切關注點
- 可測驗性
- 事件驅動
發布訂閱模式
發布者通過調度中心將訊息發送給訂閱者，調度中心解決發布與訂閱者之間的關系，保證訊息可以送達訂閱者手中，
發布者與訂閱者互不相識，發布者只管向調度中心發布訊息，而訂閱者只關心自己訂閱的訊息型別
多訂閱者保序執行
在常見的發布訂閱模式中，的確很少見到類似的說法，但在實際業務中我們會有類似的需求，一個訊息由調度中心協調多個訂閱者按照順序執行訊息，同時還可以將上一個訂閱者處理過的訊息傳遞給下一個訂閱者，這樣既可以保留發布訂閱模式的特性，又有了順序執行邏輯的特性，
一個小思考：如果 EventBus 的配置支持動態調整的話，是否業務的執行順序也可以被動態排列組合？
換句話說它或許可以為行程內作業流提供了一個可能性
Event Sourcing（事件溯源）
一種事件驅動的架構模式，可以用于審計和溯源
- 基于事件驅動架構
- 以事件為事實
- 業務資料由事件計算產生的視圖，可以持久化也可以不持久化
CQRS（命令查詢的責任分離）
CQRS 是一種架構模式，能夠使改變模型與查詢模型的實作分離
Event Sourcing & CQRS
事件溯源可以與 CQRS 很好的配合
- 在 Command Handler 中持久化事件到 Event Store 的同時實時計算一個最終視圖給 View DB 用于查詢展示
- 在 Query 中既可以通過 View DB 獲取最新狀態，也可以通過 Event Store 來重放事件來校驗 View 或用于更嚴謹的業務
Saga
Saga 是一個長活事務被分解成可以交錯運行的子事務集合，其中每個子事務都是一個保持資料庫一致性的真實事務
- 每個 Saga 由一系列 sub-transaction Ti 組成
- 每個 Ti 都有對應的補償動作 Ci，補償動作用于撤銷 Ti 造成的結果
兩種執行順序
- T1, T2, T3...[Tx retry]...,Tn
- T1, T2, ..., Tj, Cj,..., C2, C1
兩種恢復策略
- backward recovery，向后恢復，補償所有已完成的事務，如果任一子事務失敗，即上面提到的第二種執行順序，其中 j 是發生錯誤的 sub-transaction，這種做法的效果是撤銷掉之前所有成功的 sub-transation，使得整個 Saga 的執行結果撤銷
- forward recovery，向前恢復，重試失敗的事務，假設每個子事務最終都會成功，適用于必須要成功的場景，執行順序是類似于這樣的：T1, T2, ..., Tj(失敗), Tj(重試),..., Tn，其中 j 是發生錯誤的 sub-transaction，該情況下不需要 Ci
BuildingBlocks 的類視圖
作為介面標準，BuildingBlocks 中并沒有過多的干涉實作方式，它只保留了最基礎的功能流程限制，以達到最小 EventBus 的功能集合，至于最終是基于介面還是特性來實作訂閱關系的，交還給 Contrib 自行決定，
事件
用于本地事件的發布/訂閱
IEvent：事件介面，IEvent<TResult>為帶回傳值的基礎事件介面IEventHanldler<TEvent>：事件處理器介面，ISagaEventHandler<TEvent>為 Saga 的實作提供了基礎介面要求IMiddleware<TEvent>：中間件介面，允許在事件執行前掛載預處理動作和時間執行后的收尾動作IEventBus：事件總線介面，用于發送事件，并提供訂閱關系維護和附加功能執行
集成事件
用于跨行程事件的發布/訂閱
IntegrationEventLog：集成事件日志，用于實作本地訊息表的訊息模型IIntegrationEventLogService：集成事件日志服務介面ITopic：發布/訂閱的主題IIntegrationEvent：集成事件介面IIntegrationEventBus：集成事件總線，用于跨行程呼叫的事件總線
CQRS
用于使改變模型與查詢模型的實作分離
IQuery<TResult>：查詢的介面IQueryHandler<TCommand,TResult>：查詢處理器介面ICommand：可用于增刪改等指令的介面ICommandHandler<TCommand>：指令處理器介面
Event Bus
要完成上述的這些功能，我們需要借助于 EventBus，它需要有以下基礎功能
- 接收事件
- 維護訂閱關系
- 轉發事件
接收與轉發事件
這兩個功能其實可以合并為一個介面，由發布者呼叫 Publish，再由 Event Bus 根據訂閱關系轉發即可
維護訂閱關系
在.Net 專案中，我們常見的用于掃描自動注冊的方式是介面和特性
MediatR 支持介面的方式去掃描事件訂閱關系，舉個例子：IRequestHandler<,>
public class PingHandler : IRequestHandler<Ping, string>
{
public Task<string> Handle(Ping request, CancellationToken cancellationToken)
{
return Task.FromResult("Pong");
}
}
如果你的代碼潔癖程度沒有高的離譜，或許你希望是這樣
public class NetHandler : IRequestHandler<Ping, string>, IRequestHandler<Telnet, string>
{
public Task<string> Handle(Ping request, CancellationToken cancellationToken)
{
return Task.FromResult("Pong");
}
public Task<string> Handle(Telnet request, CancellationToken cancellationToken)
{
return Task.FromResult("Success");
}
}
看著好像還行？如果很多呢？
那有沒有辦法解決這個問題？
特性！我們來看個例子
public class NetHandler
{
[EventHandler]
public Task PingAsync(PingEvent @event)
{
//TODO
}
[EventHandler]
public Task TelnetAsync(TelnetEvent @event)
{
//TODO
}
}
似乎我們找到了一個出路
多訂閱者保序執行
通過事件層層推進確實可以滿足順序執行的場景，但如果你被大量無限套娃的事件包圍的時候或許你需要另外一個出路，看下例子：
public class NetHandler
{
[EventHandler(0)]
public Task PingAsync(PingEvent @event)
{
//TODO
}
[EventHandler(1)]
public Task LogAsync(PingEvent @event)
{
//TODO
}
}
只要引數是同一個 Event 就會按照 EventHandler 的 Order 順序執行，
Saga
那執行失敗了怎么辦，如果兩個方法因為其中一個需要呼叫遠程服務而無法跟本地事務結合，能幫我回滾嗎？
來吧，SAGA 走起，幫你再做個取消動作，同時還支持重試機制，以及是否忽略當前步驟的取消動作，
我們先來預設一下場景：
- 呼叫 CheckBalanceAsync 來檢查余額
- 呼叫 WithdrawAsync, 拋出 exception
- 重試 WithdrawAsync 3 次
- 呼叫 CancelWithdrawAsync
代碼如下：
public class TransferHandler
{
[EventHandler(1)]
public Task CheckBalanceAsync(TransferEvent @event)
{
//TODO
}
[EventHandler(2, FailureLevels.ThrowAndCancel, enableRetry: true, retryTimes: 3)]
public Task WithdrawAsync(TransferEvent @event)
{
//TODO
throw new Exception();
}
[EventHandler(2, FailureLevels.Ignore, enableRetry: false, isCancel: true)]
public Task CancelWithdrawAsync(TransferEvent @event)
{
//TODO
}
}
AOP
舉個業務場景，給所有 Command 在執行前增加一個引數驗證
我們提供了 Middleware，允許像俄羅斯套娃一樣（.Net Middleware）做橫切關注點的相關的事情
public class LoggingMiddleware<TEvent>
: IMiddleware<TEvent> where TEvent : notnull, IEvent
{
private readonly ILogger<LoggingMiddleware<TEvent>> _logger;
public LoggingMiddleware(ILogger<LoggingMiddleware<TEvent>> logger) => _logger = logger;
public async Task HandleAsync(TEvent @event, EventHandlerDelegate next)
{
_logger.LogInformation("----- Handling command {EventName} ({@Event})", typeof(TEvent).FullName, @event);
await next();
}
}
注冊 DI
builder.Services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>))
MASA EventBus 完整功能串列
- 接收事件
- 維護訂閱關系 - 介面
- 維護訂閱關系 - 特性
- 多訂閱者順序執行
- 轉發事件
- Saga
- AOP
- UoW
- 自動開啟和關閉事務
Integration Event Bus
用于跨服務的 Event Bus，支持最終一致性，本地訊息表
Pub/Sub
提供了 Pub Sub 介面，并基于 Dapr Pub/Sub 提供默認實作
本地訊息表
提供了本地訊息保存和 UoW 聯動介面，并基于 EF Core 提供默認實作
使用方法
啟用 Dapr Event Bus
builder.Services
.AddDaprEventBus<IntegrationEventLogService>(options=>
{
options.UseUoW<CatalogDbContext>(dbOptions => dbOptions.UseSqlServer("server=localhost;uid=sa;pwd=Password;database=test"))
.UseEventLog<CatalogDbContext>();
)
});
定義 Integration Event
public class DemoIntegrationEvent : IntegrationEvent
{
public override string Topic { get; set; } = nameof(DemoIntegrationEvent);//dapr topic name
//todo other properties
}
定義 DbContext（非必須，定義 DbContext 可以將本地訊息表與業務事務聯動）
public class CustomDbContext : IntegrationEventLogContext
{
public DbSet<User> Users { get; set; } = null!;
public CustomDbContext(MasaDbContextOptions<CustomDbContext> options) : base(options)
{
}
}
發送 Event
IIntegrationEventBus eventBus; // from DI
await eventBus.PublishAsync(new DemoIntegrationEvent());
訂閱 Event（基于 Dapr Pub/Sub 的版本）
[Topic("pubsub", nameof(DomeIntegrationEvent))]
public async Task DomeIntegrationEventHandleAsync(DomeIntegrationEvent @event)
{
//todo
}
Domain Event Bus
在領域中同時提供 Event Bus 和 Integration Event Bus 的能力，允許實時發送事件或在 Save 時一次性觸發
Domain Event Bus 是最完整的能力，所以使用 Domain Event Bus 相當于已經開啟了 Event Bus 和 Integration Event Bus，在 Domain Event Bus 內部會自動協調事件分類往 Event Bus 和 Integration Event Bus 分流
啟用 Domain Event Bus
builder.Services
.AddDomainEventBus(options =>
{
options.UseEventBus()//Use in-process events
.UseUoW<CustomDbContext>(dbOptions => dbOptions.UseSqlServer("server=localhost;uid=sa;pwd=P@ssw0rd;database=idientity"))
.UseDaprEventBus<IntegrationEventLogService>()///Use cross-process events
.UseEventLog<LocalMessageDbContext>()
.UseRepository<CustomDbContext>();
})
添加 DomainCommand
Domain Event 是行程內事件，IntegrationDomainEvent 是跨行程事件
public class RegisterUserSucceededIntegrationEvent : IntegrationDomainEvent
{
public override string Topic { get; set; } = nameof(RegisterUserSucceededIntegrationEvent);
public string Account { get; set; } = default!;
}
public class RegisterUserSucceededEvent : DomainEvent
{
public string Account { get; set; } = default!;
}
行程內事件訂閱
[EventHandler]
public Task RegisterUserHandlerAsync(RegisterUserDomainCommand command)
{
//TODO
}
跨行程事件訂閱
[Topic("pubsub", nameof(RegisterUserSucceededIntegrationEvent))]
public async Task RegisterUserSucceededHandlerAsync(RegisterUserSucceededIntegrationEvent @event)
{
//todo
}
發送 DomainCommand
IDomainEventBus eventBus;//from DI
await eventBus.PublishAsync(new RegisterUserDomainCommand());
使用場景
- 兼顧遺留系統對接
- 游走在云與非云中
- 流計算
- 微服務解耦和跨集群通信（需要將 Dapr Pub/Sub 改為 Dapr Binding，不難）
- 部分 AOP 類場景
總結
事件驅動可以解決一些特定場景的問題，凡事都有兩面性，在本來就很簡單的業務場景中使用如此復雜的模式會帶來不小的負擔，
學以致用，學無止境，
開源地址
MASA.BuildingBlocks：https://github.com/masastack/MASA.BuildingBlocks
MASA.Contrib：https://github.com/masastack/MASA.Contrib
MASA.Utils：https://github.com/masastack/MASA.Utils
MASA.EShop：https://github.com/masalabs/MASA.EShop
如果你對我們的 MASA Framework 感興趣，無論是代碼貢獻、使用、提 Issue，歡迎聯系我們
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/418977.html
標籤：C#
上一篇：出現錯誤：從Fetch更改為$http $q時，[$rootScope:inprog]$digest已經在進行中