【Shashlik.EventBus】.NET 事件總線，分布式事務最終一致性

簡介

github https://github.com/dotnet-shashlik/shashlik.eventbus

各位爺高興了給個star唄，

分布式事務、CAP定理、事件總線，在當前微服務、分布式、集群大行其道的架構前提下，是不可逃避的幾個關鍵字，在此不會過多闡述相關的理論知識，Shashlik.EventBus就是一個基于.NET6的開源事件總線解決方案，同時也是分布式事務最終一致性、延遲事件解決方案，Shashlik.EventBus采用的是異步確保的思路（本地訊息表），將訊息資料與業務資料在同一事務中進行提交或回滾，以此來保證訊息資料的可靠性，其設計目標是高性能、簡單、易用、易擴展，為拋棄歷史包袱，僅支持NET6，采用最寬松的 MIT 開源協議，

原理如下圖：

如圖所示，訊息資料需要和業務資料在同一的事務中進行提交或者回滾，最后Shashlik.EventBus會檢查訊息資料是否已提交，如果已提交才會執行真正的訊息發送，所以要求事務的隔離級別最低為讀已提交(RC)，

關于訊息冪等

Shashlik.EventBus不能保證業務訊息的冪等性，為了保證訊息的可靠傳輸，EventBus以及訊息中間件對訊息QOS處理等級必須為at least once (至少到達一次)，一般訊息中間件都需要開啟訊息持久化避免訊息丟失，簡而言之就是一個事件處理類可能處理多次同一個事件，事件訊息的冪等性應該由業務方進行處理，比如用戶訂單付款完成為一個事件，付款完成后需要修改訂單狀態為待發貨，也就是在付款完成事件處理類中可能收到多次這個訂單的付款完成事件，那么業務的冪等性處理就可以使用鎖，判斷訂單狀態，如果訂單狀態已經為待發貨，則直接回傳并忽略本次事件回應，

延遲事件

Shashlik.EventBus支持基于本地的延遲事件機制，考慮到不是所有的訊息中間件都支持延遲功能，且為了最大程度保證訊息的可靠性，最后采用了System.Timers.Timer來執行延遲功能，

延遲事件同樣適用于分布式事務最終一致性，但如果延遲事件處理類處理例外由重試器介入處理后，那么最終的延遲執行時間和期望的延遲時間就會產生較大的差異，是否忽略這里的時間差需要由具體的業務來決定，比如訂單30分鐘未付款需要關閉訂單，30分鐘后關閉訂單出現了例外，最后由重試器到了40分鐘后才關閉，也不影響訂單，那么認為這個時間差可以容忍，又比如雙11啦，發布一個延遲事件，晚上12點叫醒我起來買買買，只有1分鐘時間，過了就買不到了，那么這種情況可以在事件處理類中，自行根據當前時間、事件發送時間、延遲執行時間等要素，自行決定業務如何處理，

延遲事件和普通事件在事件定義和事件處理類宣告和處理時沒有任何區別，僅僅是在發布事件時需要指定延遲時間，

上代碼

需求：一個新用戶注冊以后有以下需求：1. 發送歡迎注冊短信；2. 發放新用戶優惠券；3. 30分鐘后推送新用戶優惠活動資訊，

1. 服務配置，這里以MySql + RabbitMQ為例：

  services.AddEventBus(r =>
      {
          // 這些都是預設配置，可以直接services.AddEventBus()
          // 運行環境，注冊到MQ的事件名稱和事件處理名稱會帶上此后綴
          r.Environment = "Production";
          // 最大失敗重試次數，默認60次
          r.RetryFailedMax = 60;
          // 訊息重試間隔，默認2分鐘
          r.RetryInterval = 60 * 2;
          // 單次重試訊息數量限制，默認100
          r.RetryLimitCount = 100;
          // 成功的訊息過期時間，默認3天，失敗的訊息永不過期，必須處理
          r.SucceedExpireHour = 24 * 3;            
          // 訊息處理失敗后，重試器介入時間，默認5分鐘后
          r.StartRetryAfter = 60 * 5;            
          // 事務提交超時時間,單位秒,默認60秒
          r.TransactionCommitTimeout = 60;
          // 重試器執行時訊息鎖定時長
          r.LockTime = 110;
      })
      // 使用ef DbContext mysql
      .AddMySql<DemoDbContext>()
      // 配置RabbitMQ
      .AddRabbitMQ(r =>
      {
          r.Host = "localhost";
          r.UserName = "rabbit";
          r.Password = "123123";
      });

2. 定義事件

  // 新用戶注冊完成事件，實作介面IEvent
  public class NewUserEvent : IEvent
  {
      public string Id { get;set; }
      public string Name { get; set; }
  }
  
  // 定義新用戶注冊延遲活動推送事件
  public class NewUserPromotionEvent : IEvent
  {
      public string Id { get;set; }
      public string Name { get; set; }
      public string PromotionId { get; set; }
  }

3. 發布事件

  public class UserManager
  {
      public UserManager(IEventPublisher eventPublisher, DemoDbContext dbContext)
      {
          EventPublisher = eventPublisher;
          DbContext = dbContext;
      }
  
      private IEventPublisher EventPublisher { get; }
      private DemoDbContext DbContext { get; }
  
      public async Task CreateUserAsync(UserInput input)
      {
          // 開啟本地事務
          using var tran = await DbContext.DataBase.BeginTransactionAsync();
          try
          {
              // 創建用戶邏輯處理...
  
              // 發布新用戶事件
              // 通過注入IEventPublisher發布事件，需要傳入事務背景關系資料
              await EventPublisher.PublishAsync(new NewUserEvent{
                  Id = user.Id,
                  Name = input.Name
              }, DbContext.GetTransactionContext());
  
              // 發布延遲事件
              // 通過ef擴展，直接使用DbContext發布事件，自動使用當前背景關系事務
              await DbContext.PublishEventAsync(new NewUserPromotionEvent{
                  Id = user.Id,
                  Name = input.Name,
                  PromotionId = "1"
              }, DatetimeOffset.Now.AddMinutes(30));
  
              // 提交本地事務
              await tran.CommitAsync();
          }catch(Exception ex)
          {
              // 回滾事務，訊息資料也將回滾不會發布
              await tran.RollbackAsync();
          }
      }
  }

4. 定義事件處理類

    
  // 一個事件可以有多個處理類，可以分布在不同的微服務中
  // 用于發送短信的事件處理類
  public class NewUserEventForSmsHandler : IEventHandler<NewUserEvent>
  {
      public async Task Execute(NewUserEvent @event, IDictionary<string, string> items)
      {
          // 發送短信...
      }
  }


  // 用于發放消費券的事件處理類
  public class NewUserEventForCouponsHandler : IEventHandler<NewUserEvent>
  {
      public async Task Execute(NewUserEvent @event, IDictionary<string, string> items)
      {
          // 業務處理...
      }
  }

  // 用于新用戶延遲活動的事件處理類,將在指定時間執行
  public class NewUserPromotionEventHandler : IEventHandler<NewUserPromotionEvent>
  {
      public async Task Execute(NewUserPromotionEvent @event, IDictionary<string, string> items)
      {
          // 業務處理...
      }
  }    

默認的，發布、宣告到訊息中間件的事件、事件處理器名稱生產規則為{Type.Name}.{Options.Environment}，在分布式架構下需要，您需要了解這個默認規則，這點不同于CAP框架必須顯示宣告，當然Shashlik.EventBus也可以使用EventBusNameAttribute特性來顯示宣告，詳細說明請上github查看wiki檔案，

XA事務支持（TransactionScope）

雖然盡可能的不要使用TransactionScope，但在某些場景仍然是需要的，Shashlik.EventBus對其提供了事務支持，可以通過XaTransactionContext.Current獲取當前環境的事務背景關系，發布事件如下：

  public class UserManager
  {
      public UserManager(IEventPublisher eventPublisher, DemoDbContext dbContext)
      {
          EventPublisher = eventPublisher;
          DbContext = dbContext;
      }
  
      private IEventPublisher EventPublisher { get; }
      private DemoDbContext DbContext { get; }
  
      public async Task CreateUserAsync(UserInput input)
      {
          // 開啟事務
          using var scope = new TransactionScope();
          try
          {
              // 創建用戶邏輯處理...
  
              // 發布新用戶事件
              // 通過注入IEventPublisher發布事件，需要傳入事務背景關系資料
              await EventPublisher.PublishAsync(new NewUserEvent{
                  Id = user.Id,
                  Name = input.Name
              // 使用 XaTransactionContext.Current
              }, XaTransactionContext.Current);
  
              // 提交事務
              await scope.Complete();
          }catch(Exception ex)
          {
              // 回滾事務，訊息資料也將回滾不會發布
              await tran.RollbackAsync();
          }
      }
  }

擴展

如果默認實作不能滿足你的需求，可以自行實作可擴展介面，并注冊即可，

• IMsgIdGenerator：訊息Id生成器，是指傳輸的全域唯一id，不是指存盤的id，默認guid
• IEventPublisher：事件發布處理器，
• IMessageSerializer：訊息序列化、反序列化處理類，默認Newtonsoft.Json，
• IReceivedMessageRetryProvider：已接收訊息重試器，
• IPublishedMessageRetryProvider：已發布訊息重試器，
• IEventHandlerInvoker: 事件處理執行器
• IEventNameRuler：事件名稱規則生成(對應訊息佇列topic/route)，
• IEventHandlerNameRuler：事件處理名稱規則生成(對應訊息佇列queue/group)，
• IEventHandlerFindProvider：事件處理類查找器
• IExpiredMessageProvider：已過期訊息洗掉處理器，
• IMessageListener：訊息監聽處理器，
• IRetryProvider：重試執行器，
• IPublishHandler：訊息發布處理器，
• IReceivedHandler：訊息接收處理器，
• IMessageStorageInitializer：存盤介質初始化，
• IMessageStorage：訊息存盤、讀取等操作，

例：

  // 替換默認的IMsgIdGenerator
  service.AddSingleton<IMsgIdGenerator, CustomMsgIdGenerator>();
  service.AddEventBus()
      .AddMemoryQueue()
      .AddMemoryStorage();

后續計劃

• 功能
  • Dashboard
• 訊息中間件支持
  • RabbitMQ
  • Kafka
  • RocketMQ
  • ActiveMQ
  • Pulsar
  • Redis
• 存盤支持
  • MySql
  • PostgreSql
  • SqlServer
  • Oracle
  • MongoDb

標籤：.NET Core

上一篇：讀 MAUI 源代碼 理解可系結物件和可系結屬性的存盤機制

下一篇：讀 MAUI 源代碼 理解可系結物件和可系結屬性的存盤機制