MASA Framework - EventBus 設計

概述

利用發佈訂閱模式來解耦不同架構層級，亦可用於解決隔離業務之間的交互

優點：

松耦合
橫切關注點
可測試性
事件驅動

發佈訂閱模式

發佈者通過調度中心將消息發送給訂閱者。調度中心解決發佈與訂閱者之間的關係，保證消息可以送達訂閱者手中。

發佈者與訂閱者互不相識，發佈者只管向調度中心發佈消息，而訂閱者只關心自己訂閱的消息類型

多訂閱者保序執行

在常見的發佈訂閱模式中，的確很少見到類似的說法。但在實際業務中我們會有類似的需求，一個消息由調度中心協調多個訂閱者按照順序執行消息，同時還可以將上一個訂閱者處理過的消息傳遞給下一個訂閱者。這樣既可以保留髮布訂閱模式的特性，又有了順序執行邏輯的特性。

一個小思考：如果 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，歡迎聯繫我們

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/VKyprsQAcKXO1k-nVgGveQ

概述

發佈訂閱模式

多訂閱者保序執行

Event Sourcing（事件溯源）

CQRS（命令查詢的責任分離）

Event Sourcing & CQRS

Saga

BuildingBlocks 的類視圖

事件

集成事件

CQRS

Event Bus

接收與轉發事件

維護訂閱關係

多訂閱者保序執行

Saga

AOP

MASA EventBus 完整功能列表

Integration Event Bus

使用方法

Domain Event Bus

使用場景

總結

開源地址

猜你喜歡