目錄

MASA Framework - 整體設(shè)計(jì)思路
MASA Framework - EventBus設(shè)計(jì)
MASA Framework - MASA Framework - DDD設(shè)計(jì)（1）
MASA Framework - MASA Framework - DDD設(shè)計(jì)（2）

概述

利用發(fā)布訂閱模式來(lái)解耦不同架構(gòu)層級(jí)，亦可用于解決隔離業(yè)務(wù)之間的交互

優(yōu)點(diǎn)：

• 松耦合
• 橫切關(guān)注點(diǎn)
• 可測(cè)試性
• 事件驅(qū)動(dòng)

發(fā)布訂閱模式

發(fā)布者通過(guò)調(diào)度中心將消息發(fā)送給訂閱者。調(diào)度中心解決發(fā)布與訂閱者之間的關(guān)系，保證消息可以送達(dá)訂閱者手中。

發(fā)布者與訂閱者互不相識(shí)，發(fā)布者只管向調(diào)度中心發(fā)布消息，而訂閱者只關(guān)心自己訂閱的消息類型

多訂閱者保序執(zhí)行

在常見(jiàn)的發(fā)布訂閱模式中，的確很少見(jiàn)到類似的說(shuō)法。但在實(shí)際業(yè)務(wù)中我們會(huì)有類似的需求，一個(gè)消息由調(diào)度中心協(xié)調(diào)多個(gè)訂閱者按照順序執(zhí)行消息，同時(shí)還可以將上一個(gè)訂閱者處理過(guò)的消息傳遞給下一個(gè)訂閱者。這樣既可以保留發(fā)布訂閱模式的特性，又有了順序執(zhí)行邏輯的特性。

一個(gè)小思考：如果EventBus的配置支持動(dòng)態(tài)調(diào)整的話，是否業(yè)務(wù)的執(zhí)行順序也可以被動(dòng)態(tài)排列組合？

換句話說(shuō)它或許可以為進(jìn)程內(nèi)工作流提供了一個(gè)可能性

Event Sourcing（事件溯源）

一種事件驅(qū)動(dòng)的架構(gòu)模式，可以用于審計(jì)和溯源

• 基于事件驅(qū)動(dòng)架構(gòu)
• 以事件為事實(shí)
• 業(yè)務(wù)數(shù)據(jù)由事件計(jì)算產(chǎn)生的視圖，可以持久化也可以不持久化

CQRS（命令查詢的責(zé)任分離）

CQRS是一種架構(gòu)模式，能夠使改變模型與查詢模型的實(shí)現(xiàn)分離

Event Sourcing & CQRS

事件溯源可以與CQRS很好的配合

• 在Command Handler中持久化事件到Event Store的同時(shí)實(shí)時(shí)計(jì)算一個(gè)最終視圖給View DB用于查詢展示
• 在Query中既可以通過(guò)View DB獲取最新狀態(tài)，也可以通過(guò)Event Store來(lái)重放事件來(lái)校驗(yàn)View或用于更嚴(yán)謹(jǐn)?shù)臉I(yè)務(wù)

Saga

Saga是一個(gè)長(zhǎng)活事務(wù)被分解成可以交錯(cuò)運(yùn)行的子事務(wù)集合。其中每個(gè)子事務(wù)都是一個(gè)保持?jǐn)?shù)據(jù)庫(kù)一致性的真實(shí)事務(wù)

• 每個(gè)Saga由一系列sub-transaction Ti 組成
• 每個(gè)Ti 都有對(duì)應(yīng)的補(bǔ)償動(dòng)作Ci，補(bǔ)償動(dòng)作用于撤銷Ti造成的結(jié)果

兩種執(zhí)行順序

• T1, T2, T3...[Tx retry]...,Tn
• T1, T2, ..., Tj, Cj,..., C2, C1

兩種恢復(fù)策略

• backward recovery，向后恢復(fù)，補(bǔ)償所有已完成的事務(wù)，如果任一子事務(wù)失敗。即上面提到的第二種執(zhí)行順序，其中j是發(fā)生錯(cuò)誤的sub-transaction，這種做法的效果是撤銷掉之前所有成功的sub-transation，使得整個(gè)Saga的執(zhí)行結(jié)果撤銷
• forward recovery，向前恢復(fù)，重試失敗的事務(wù)，假設(shè)每個(gè)子事務(wù)最終都會(huì)成功。適用于必須要成功的場(chǎng)景，執(zhí)行順序是類似于這樣的：T1, T2, ..., Tj(失敗), Tj(重試),..., Tn，其中j是發(fā)生錯(cuò)誤的sub-transaction。該情況下不需要Ci

BuildingBlocks的類視圖

作為接口標(biāo)準(zhǔn)，BuildingBlocks中并沒(méi)有過(guò)多的干涉實(shí)現(xiàn)方式，它只保留了最基礎(chǔ)的功能流程限制，以達(dá)到最小EventBus的功能集合。至于最終是基于接口還是特性來(lái)實(shí)現(xiàn)訂閱關(guān)系的，交還給Contrib自行決定。

事件

用于本地事件的發(fā)布/訂閱

• IEvent：事件接口，IEvent<TResult>為帶返回值的基礎(chǔ)事件接口
• IEventHanldler<TEvent>：事件處理器接口，ISagaEventHandler<TEvent>為Saga的實(shí)現(xiàn)提供了基礎(chǔ)接口要求
• IMiddleware<TEvent>：中間件接口，允許在事件執(zhí)行前掛載預(yù)處理動(dòng)作和時(shí)間執(zhí)行后的收尾動(dòng)作
• IEventBus：事件總線接口，用于發(fā)送事件，并提供訂閱關(guān)系維護(hù)和附加功能執(zhí)行

集成事件

用于跨進(jìn)程事件的發(fā)布/訂閱

• IntegrationEventLog：集成事件日志，用于實(shí)現(xiàn)本地消息表的消息模型
• IIntegrationEventLogService：集成事件日志服務(wù)接口
• ITopic：發(fā)布/訂閱的主題
• IIntegrationEvent：集成事件接口
• IIntegrationEventBus：集成事件總線，用于跨進(jìn)程調(diào)用的事件總線

CQRS

用于使改變模型與查詢模型的實(shí)現(xiàn)分離

• IQuery<TResult>：查詢的接口
• IQueryHandler<TCommand,TResult>：查詢處理器接口
• ICommand：可用于增刪改等指令的接口
• ICommandHandler<TCommand>：指令處理器接口

Event Bus

要完成上述的這些功能，我們需要借助于EventBus，它需要有以下基礎(chǔ)功能

• 接收事件
• 維護(hù)訂閱關(guān)系
• 轉(zhuǎn)發(fā)事件

接收與轉(zhuǎn)發(fā)事件

這兩個(gè)功能其實(shí)可以合并為一個(gè)接口，由發(fā)布者調(diào)用Publish，再由Event Bus根據(jù)訂閱關(guān)系轉(zhuǎn)發(fā)即可

維護(hù)訂閱關(guān)系

在.Net項(xiàng)目中，我們常見(jiàn)的用于掃描自動(dòng)注冊(cè)的方式是接口和特性

MediatR支持接口的方式去掃描事件訂閱關(guān)系，舉個(gè)例子：IRequestHandler<,>

public class PingHandler : IRequestHandler<Ping, string>
{
    public Task<string> Handle(Ping request, CancellationToken cancellationToken)
    {
        return Task.FromResult("Pong");
    }
}

如果你的代碼潔癖程度沒(méi)有高的離譜，或許你希望是這樣

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");
    }
}

看著好像還行？如果很多呢？

那有沒(méi)有辦法解決這個(gè)問(wèn)題？

特性！我們來(lái)看個(gè)例子

public class NetHandler
{
    [EventHandler]
    public Task PingAsync(PingEvent @event)
    {
        //TODO
    }
    
    [EventHandler]
    public Task TelnetAsync(TelnetEvent @event)
    {
        //TODO
    }
}

似乎我們找到了一個(gè)出路

多訂閱者保序執(zhí)行

通過(guò)事件層層推進(jìn)確實(shí)可以滿足順序執(zhí)行的場(chǎng)景，但如果你被大量無(wú)限套娃的事件包圍的時(shí)候或許你需要另外一個(gè)出路，看下例子：

public class NetHandler
{
    [EventHandler(0)]
    public Task PingAsync(PingEvent @event)
    {
        //TODO
    }
    
    [EventHandler(1)]
    public Task LogAsync(PingEvent @event)
    {
        //TODO
    }
}

只要參數(shù)是同一個(gè)Event就會(huì)按照EventHandler的Order順序執(zhí)行。

Saga

那執(zhí)行失敗了怎么辦，如果兩個(gè)方法因?yàn)槠渲幸粋€(gè)需要調(diào)用遠(yuǎn)程服務(wù)而無(wú)法跟本地事務(wù)結(jié)合，能幫我回滾嗎？

來(lái)吧，SAGA走起，幫你再做個(gè)取消動(dòng)作，同時(shí)還支持重試機(jī)制，以及是否忽略當(dāng)前步驟的取消動(dòng)作。

我們先來(lái)預(yù)設(shè)一下場(chǎng)景：

1. 調(diào)用CheckBalanceAsync來(lái)檢查余額
2. 調(diào)用WithdrawAsync, 拋出 exception
3. 重試WithdrawAsync 3 次
4. 調(diào)用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

舉個(gè)業(yè)務(wù)場(chǎng)景，給所有Command在執(zhí)行前增加一個(gè)參數(shù)驗(yàn)證

我們提供了Middleware，允許像俄羅斯套娃一樣（.Net Middleware）做橫切關(guān)注點(diǎn)的相關(guān)的事情

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();
    }
}

注冊(cè)DI

builder.Services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>))

MASA EventBus完整功能列表

• 接收事件
• 維護(hù)訂閱關(guān)系 - 接口
• 維護(hù)訂閱關(guān)系 - 特性
• 多訂閱者順序執(zhí)行
• 轉(zhuǎn)發(fā)事件
• Saga
• AOP
• UoW
• 自動(dòng)開啟和關(guān)閉事務(wù)

Integration Event Bus

用于跨服務(wù)的Event Bus，支持最終一致性，本地消息表

Pub/Sub

提供了Pub Sub接口，并基于Dapr Pub/Sub提供默認(rèn)實(shí)現(xiàn)

本地消息表

提供了本地消息保存和UoW聯(lián)動(dòng)接口，并基于EF Core提供默認(rèn)實(shí)現(xiàn)

使用方法

啟用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可以將本地消息表與業(yè)務(wù)事務(wù)聯(lián)動(dòng)）

public class CustomDbContext : IntegrationEventLogContext
{
    public DbSet<User> Users { get; set; } = null!;

    public CustomDbContext(MasaDbContextOptions<CustomDbContext> options) : base(options)
    {

    }
}

發(fā)送 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

在領(lǐng)域中同時(shí)提供Event Bus和Integration Event Bus的能力，允許實(shí)時(shí)發(fā)送事件或在Save時(shí)一次性觸發(fā)

Domain Event Bus是最完整的能力，所以使用Domain Event Bus相當(dāng)于已經(jīng)開啟了Event Bus和Integration Event Bus，在Domain Event Bus內(nèi)部會(huì)自動(dòng)協(xié)調(diào)事件分類往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是進(jìn)程內(nèi)事件，IntegrationDomainEvent是跨進(jìn)程事件

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!;
}

進(jìn)程內(nèi)事件訂閱

[EventHandler]
public Task RegisterUserHandlerAsync(RegisterUserDomainCommand command)
{
    //TODO
}

跨進(jìn)程事件訂閱

[Topic("pubsub", nameof(RegisterUserSucceededIntegrationEvent))]
public async Task RegisterUserSucceededHandlerAsync(RegisterUserSucceededIntegrationEvent @event)
{
    //todo
}

發(fā)送DomainCommand

IDomainEventBus eventBus;//from DI
await eventBus.PublishAsync(new RegisterUserDomainCommand());

使用場(chǎng)景

• 兼顧遺留系統(tǒng)對(duì)接
• 游走在云與非云中
• 流計(jì)算
• 微服務(wù)解耦和跨集群通信（需要將Dapr Pub/Sub改為Dapr Binding，不難）
• 部分AOP類場(chǎng)景

總結(jié)

事件驅(qū)動(dòng)可以解決一些特定場(chǎng)景的問(wèn)題，凡事都有兩面性，在本來(lái)就很簡(jiǎn)單的業(yè)務(wù)場(chǎng)景中使用如此復(fù)雜的模式會(huì)帶來(lái)不小的負(fù)擔(dān)。

學(xué)以致用，學(xué)無(wú)止境。

開源地址

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

如果你對(duì)我們的MASA Framework感興趣，無(wú)論是代碼貢獻(xiàn)、使用、提Issue，歡迎聯(lián)系我們