MASA Framework - EventBus设计

概述

利用发布订阅模式来解耦不同架构层级，亦可用于解决隔离业务之间的交互

优点：

松耦合
横切关注点
可测试性
事件驱动

发布订阅模式

发布者通过调度中心将消息发送给订阅者。调度中心解决发布与订阅者之间的关系，保证消息可以送达订阅者手中。

发布者与订阅者互不相识，发布者只管向调度中心发布消息，而订阅者只关心自己订阅的消息类型

event bus design.png

多订阅者保序执行

在常见的发布订阅模式中，的确很少见到类似的说法。但在实际业务中我们会有类似的需求，一个消息由调度中心协调多个订阅者按照顺序执行消息，同时还可以将上一个订阅者处理过的消息传递给下一个订阅者。这样既可以保留发布订阅模式的特性，又有了顺序执行逻辑的特性。

一个小思考：如果EventBus的配置支持动态调整的话，是否业务的执行顺序也可以被动态排列组合？

换句话说它或许可以为进程内工作流提供了一个可能性

event bus design - keep order.png

Event Sourcing（事件溯源）

一种事件驱动的架构模式，可以用于审计和溯源

基于事件驱动架构
以事件为事实
业务数据由事件计算产生的视图，可以持久化也可以不持久化

CQRS（命令查询的责任分离）

CQRS是一种架构模式，能够使改变模型与查询模型的实现分离

Event Sourcing & CQRS

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

在Command Handler中持久化事件到Event Store的同时实时计算一个最终视图给View DB用于查询展示
在Query中既可以通过View DB获取最新状态，也可以通过Event Store来重放事件来校验View或用于更严谨的业务

event sourcing cqrs.png

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为带返回值的基础事件接口
IEventHanldler：事件处理器接口，ISagaEventHandler为Saga的实现提供了基础接口要求
IMiddleware：中间件接口，允许在事件执行前挂载预处理动作和时间执行后的收尾动作
IEventBus：事件总线接口，用于发送事件，并提供订阅关系维护和附加功能执行

events code map.png

集成事件

用于跨进程事件的发布/订阅

IntegrationEventLog：集成事件日志，用于实现本地消息表的消息模型
IIntegrationEventLogService：集成事件日志服务接口
ITopic：发布/订阅的主题
IIntegrationEvent：集成事件接口
IIntegrationEventBus：集成事件总线，用于跨进程调用的事件总线

integration event code map.png

CQRS

用于使改变模型与查询模型的实现分离

IQuery：查询的接口
IQueryHandler：查询处理器接口
ICommand：可用于增删改等指令的接口
ICommandHandler：指令处理器接口

cqrs code map.png

Event Bus

要完成上述的这些功能，我们需要借助于EventBus，它需要有以下基础功能

接收事件
维护订阅关系
转发事件

接收与转发事件

这两个功能其实可以合并为一个接口，由发布者调用Publish，再由Event Bus根据订阅关系转发即可

维护订阅关系

在.Net项目中，我们常见的用于扫描自动注册的方式是接口和特性

MediatR支持接口的方式去扫描事件订阅关系，举个例子：IRequestHandler<,>

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

如果你的代码洁癖程度没有高的离谱，或许你希望是这样

public class NetHandler : IRequestHandler, IRequestHandler
{
    public Task Handle(Ping request, CancellationToken cancellationToken)
    {
        return Task.FromResult("Pong");
    }
    
    public Task 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
    : IMiddleware where TEvent : notnull, IEvent
{
    private readonly ILogger> _logger;
        
    public LoggingMiddleware(ILogger> 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(options=>
    {
        options.UseUoW(dbOptions => dbOptions.UseSqlServer("server=localhost;uid=sa;pwd=Password;database=test"))
               .UseEventLog();
        )
    });

定义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 Users { get; set; } = null!;

    public CustomDbContext(MasaDbContextOptions 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(dbOptions => dbOptions.UseSqlServer("server=localhost;uid=sa;pwd=P@ssw0rd;database=idientity"))
        .UseDaprEventBus()///Use cross-process events
        .UseEventLog()
        .UseRepository();
})

添加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，欢迎联系我们

Framework MasaFramework

MASA Framework - EventBus设计

目录

概述

发布订阅模式

多订阅者保序执行

Event Sourcing（事件溯源）

CQRS（命令查询的责任分离）

Event Sourcing & CQRS

Saga

BuildingBlocks的类视图

事件

集成事件

CQRS

Event Bus

接收与转发事件

维护订阅关系

多订阅者保序执行

Saga

AOP

MASA EventBus完整功能列表

Integration Event Bus

使用方法

Domain Event Bus

使用场景

总结

开源地址

相关

通俗易懂，什么是.NET?什么是.NET Framework？什么是.NET Core?

Django Rest Framework的使用

org.springframework.data.redis.RedisConnectionFailureException

Python3+Robot Framework环境搭建

CefSharp基于.Net Framework 4.0 框架编译

django rest framework

SpringMVC中org.springframework.web.servlet.DispatcherServlet报红

Entity Framework 6.x Code Frist For Oracle 实践与注意点

【Django学习】Django Rest Framework 的使用

windows2016安装.netFramework 3.5

net core读取文件编码类型和 net framework区别

Frameworks Detected:Web framework is detected.

标签