Abp vNnext-事件总线使用实现及解析
事件总线的基本使用1.引入模块AbpEventBusModule模块
2.注入本地事件发布接口 ,以本地事件总线举例, 因为思路都差不多,但是分布式事件的稍微配置麻烦一些
3.先定义事件传输数据结构
public class StockCountChangedEto
{
public Guid ProductId { get; set; }
}4.定义事件处理程序
//分布式事件订阅自IDistributedEventHandler<>
//public class MyEventHandler:IDistributedEventHandler<StockCountChangedEto>,ITransientDependency
public class MyEventHandler:ILocalEventHandler<StockCountChangedEto>,ITransientDependency
{
public async Task HandleEventAsync(StockCountChangedEto eventData)
{
//todo somthing
}
}
//手动注册
_eventBus.Subscribe<StockCountChangedEto>(new MyEventHandler());5.如果你有在代码上下文订阅事件的需求
LocalEventBus.Subscribe<MySimpleEventData>(eventData =>
{
totalData += eventData.Value;
return Task.CompletedTask;
});6.发布事件
private readonly ILocalEventBus _eventBus;
var publishEto = new StockCountChangedEto{ ProductId = Guid.NewGuid()}
await _eventBus.PublishAsync(publishEto , false);
// 分布式事件总线发布
//private readonly IDistributedEventBus _distributeEventBus;
// var publishEto1 = new StockCountChangedEto{ ProductId = Guid.NewGuid()}
// await _distributeEventBus.PublishAsync(publishEto , false);实现分析
1.基本概念
在这之前我们要搞清楚3个重要的角色 (订阅者,发布者,消息事件)以及他们的关系,我们做这件事的流程就是 发布- 事件消息- 到订阅者
[*]事件（Event）：表示系统中发生的事情或状态变化。事件可以携带有关该事件的数据。
[*]发布者（Publisher）生产者（Producer）：产生事件的组件。它们不直接与事件的消费者交互，而是通过事件总线发送事件。
[*]订阅者（Subscriber）消费者（Consumer）：对特定类型的事件感兴趣的组件。当事件发生时，如果事件类型匹配，事件总线会通知相应的订阅者。
[*]事件总线（Event Bus）：中介者角色，负责管理事件的订阅、取消订阅以及事件的分发工作。
2.如何理解事件总线的功能
从需求上来说就是我需要发布,然后有个订阅,最简单的观察者模式,例如可以通过mq redis这些来发送 , 从大方向来看,他的设计很简单.
1.内部维护一个事件订阅的 源数据字典 , key 是消息数据的类型, values是具体的处理程序Handler
2.在发布时根据发送消息的数据类型,去匹配可用的事件源,然后依次次调用每个Handler
不过abp除了实现主体功能外, 还把这个需求落地为一个通用可扩展的功能,接下来我们看看内部是如何一步步实现的
1.理解他的设计
2.学习设计思想
3.学习他的一些代码写法
尝试理解它内部实现的原理和设计思想, 一开始我的思路是 从发送事件开始一步步跟逻辑看,光看,但是发现效果不好,只能知道他是怎么走的, 过一段时间忘了, 理解的知识没有结构化, 梳理了一
阵之后, 才发现还是得,分步了解他的思路, 这里用本地事件来作为主要的对象解读。
1.如何实现订阅，订阅的方法
2.如何发布，结构参与的类及结构，执行流程
先从订阅开始，为啥呢，因为你只有订阅了才有地方发啊, 但是正常在一接触的时候应该和我开始差不多,流水账式的读，读完了，跟没读一样，所以经过了几次之后发现这部分,从订阅开始着手,
才可能更好理解。
3.订阅
事件总线为我们提供了2种订阅的方式。
关于订阅的类图如下：
3.1.(方式一) 启动时订阅
ABP 框架为事件处理抽象了一个顶级的标记接口 IEventHandler。它本身没有成员，主要用于标识一个类是事件处理程序。
机制：
[*]在应用模块加载时，通过依赖注入系统自动发现并注册所有实现了特定事件处理接口的类。
要求：
[*]本地事件处理程序必须实现 ILocalEventHandler 接口。
[*]分布式事件处理程序必须实现 IDistributedEventHandler 接口。
[*]这两个接口都继承自 IEventHandler。
public class MySimpleEventDataHandler : ILocalEventHandler<MySimpleEventData>, ISingletonDependency
{
public int TotalData { get; private set; }
public Task HandleEventAsync(MySimpleEventData eventData)
{
TotalData += eventData.Value;
return Task.CompletedTask;
}
}3.2.(方式二) 直接调用api订阅
通过事件总线提供的 Subscribe() 来在代码中进行动态订阅,这种方式非常灵活，支持多种订阅形式，可以提供委托,也可以自己直接构造传入Handler.
// 通过委托实现1 ,处理器单例周期
LocalEventBus.Subscribe<MySimpleEventData>(eventData =>
{
// 执行订阅的代码
totalData += eventData.Value;
return Task.CompletedTask;
});
// api实现2,处理器瞬时周期
LocalEventBus.Subscribe<MySimpleEventData, MySimpleTransientEventHandler>();
// 直接构造实现3,处理器单例周期
var handler = new MyEventHandler();
LocalEventBus.Subscribe<EntityChangedEventData<MyEntity>>(handler);
// 发布
await LocalEventBus.PublishAsync(new MySimpleEventData(1));3.3.内部如何实现2种订阅
如果您平时接触或者实际编码使用过abp的事件总线，应该不会太陌生，可以接着往下看，无论是条件订阅，还是代码行内实现，在这2种方式最终都是存在一个地方,这里画了图，看图理解。
不过这里图是上面整体简化的订阅部分,图给出的是大致思路,具体实现肯定还是经过一系列的处理和扩展的，在后续会分别对内部详细的过程进行分析的。
3.3.1.模块加载时注册 (配置选项 + abp服务注册事件回调)
1.框架定义了2个选项分别是 AbpLocalEventBusOptions和 AbpDistributedEventBusOptions 在模块加载时的服务注册事件回调中,会使用反射扫描程序集中分别实现了2个泛
型接口的类,然后把他们加入到选项中的集合类型里.
然后在事件总线核心类 LocalEventBus 构造时, 将选项注入到类中，然后调用 SubscribeHandlers() 进行初始订阅。
2.SubscribeHandlers() 的主要逻辑就是在最终注册前再一次进行校验,校验逻辑如下.
1.必须实现自IEventHandler其实只要handler类实现了 ILocalEventHandler就行,因为 框架中默认将ILocalEventHandler 和IDistributedEventHandler 都实现自
IEventHandler, 突然发现框架中这个代码有点多余.
2.实现ILocalEventHandler必须包含一个泛型参数,如果没有参数也是不能订阅的.
3.校验通过后调用重载方法传入指定参数, 第一个是泛型的具体类型(就是事件对象) ,第二个参数是把handler处理器使用一个实现自IEventHandlerFactory的 IocEventHandlerFactory 类包装了一下.
Subscribe(genericArgs, new IocEventHandlerFactory(ServiceScopeFactory, handler));4.其实最终订阅的方法是调用类中另外一个重载的方法Subscribe , LocalEventBus在构造时内部维护了一个线程安全的并发缓存字典，它的作用就是将订阅的信息加入到处理器工厂列
表，我按照自己的理解给他一个定义就叫事件订阅缓存映射列表,后面就用它来描述， 至此就完成了注册.
[*]key:是消息事件的类型
[*]value是用IEventHandlerFactory包裹的handler
// 缓存字典,key:是消息事件的类型,value是用IEventHandlerFactory包裹的handler
protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
{
// 往HandlerFactories 里加入当前事件对象类型
// 加锁判断是否在处理器工厂列表中存在,不存在就加入
GetOrCreateHandlerFactories(eventType)
.Locking(factories =>
{
if (!factory.IsInFactories(factories))
{
factories.Add(factory);
}
}
);
return new EventHandlerFactoryUnregistrar(this, eventType, factory);
}5.这里要展开说一下订阅时为何要将Handler用一个IocEventHandlerFactory 类包装一下, 它是实现自IEventHandlerFactory,先看下面的类图关系 .
IEventHandlerFactory有3个具体实现，看名字就能猜出跟生命周期有关系，它的存在使不同的注册方式之间获取到的处理器实例的生命周期有一些差别，如果业务中有对处理器执行时生命周期
的要求，可以选择不同的api来实现，具体看下面代码。
[*]IocEventHandlerFactory:从 IoC 容器解析处理器实例,例如通过模块加载时配置注册的都是从容器解析处理器作用域实例.
[*]SingleInstanceHandlerFactory:使用预先创建的处理器实例,以下代码注册的处理器就是单例周期
var handler = new MyEventHandler();
LocalEventBus.Subscribe<EntityChangedEventData<MyEntity>>(handler);
[*]TransientEventHandlerFactory:每次调用都创建新的处理器实例,以下代码注册的处理器就是瞬时周期
LocalEventBus.Subscribe<MySimpleEventData, MySimpleTransientEventHandler>();咱们再关注接口IEventHandlerFactory中具体提供的2个方法
一个是获取事件处理器的GetHandler()大概可以猜出它的作用，就是在发布事件后，根据具体的事件对象类型，在已有的事件订阅缓存映射列表中，获取指定的处理handler。
一个是IsInFactories() 在新增时进行重复注册检查, 判断处理器在不在 事件订阅缓存映射列表中的，如果不在就加入事件订阅缓存映射列表。
3.3.3.使用Api注册
使用api的方法有好几种,拿一个最特殊的使用委托注册来说,它是直接写在代码上下文里面的,这个时候你可能会有疑问,事件处理器的约定不是本地事件订阅必须实现ILocalEventHandler
或者分布式事件必须实现IDistributedEventHandler吗？直接写怎么弄的
// 通过委托实现1 ,处理器单例周期
LocalEventBus.Subscribe<MySimpleEventData>(eventData =>
{
// 执行订阅的代码
totalData += eventData.Value;
return Task.CompletedTask;
});其实对于委托订阅的方式，框架默认定义了了一个ActionEventHandler 也实现自ILocalEventHandler，后续的订阅流程和上面一毛一样的，就不废话啦。
4.发布
4.1.基础结构
接着看发布是如何实现的，发布是使用 PublishAsync(TEvent eventData) 来完成，他最少允许接收一个参数，而这个参数就是事件的消息对象，大白话就是你需要发送给订阅者的数据，发布
消息的通用Api基本都定义在抽象类 EventBusBase 中，但是最终发布还是由实现了抽象类的具体实现类来发送的，例如本地事件由 LocalEventBus 类来承担**执行发送的。
分布式事件相对于本地稍微特殊一点，看下面类图
[*]第一层红色标记的 IEventBus 接口是整个事件总线发布对象的抽象。
[*]第2层黄色标记的 ILocalEventBus 和 IDistributedEventBus 接口是本地事件和分布式事件的接口抽象。
[*]第3层绿色标记的 LocalEventBus 是本地事件的直接发布对象，他继承自 EventBusBase ，实现 ILocalEventBus ，关于本地事件的所有是由它来执行，并且结构层级到这一层就结束了。与它平级的就是关于分布式事件的抽象类 DistributedEventBusBase
[*]第4层无颜色标记的 RabbitMqDistributedEventBus 和 KafkaDistributedEventBus 它们的作用就不多赘述，顾名思义，它们都是实现自 DistributedEventBusBase
，也就是说再扩展分布式相关的事件总线只需要同样继承就行了，当然框架还提供了其他几种，如果感兴趣的，可以按照这个标准，自己利用Redis的发布订阅来实现一套，因为框架没有提供利用redis 作为作为事件总线的实现。
4.2.问题疑问（重点）
为什么需要 DistributedEventBusBase这层抽象？
在理解 ABP 事件总线设计时，一个核心问题是：为什么分布式事件总线不像 LocalEventBus一样直接继承 EventBusBase，而是要额外抽象出一个 DistributedEventBusBase基类
然后由例如 KafkaDistributedEventBus 、RabbitMqDistributedEventBus 来实现，脑子里隐约感觉知道为什么，但是久久说不上来，也总结不出来，我相信很多小伙伴都有这样的情景，不过好在不知道答案没关系，能发现关键问题也不错，后面经过和DeepSeek的深入交流，我觉得它总结的相当到位，如下：
1.截然不同的职责
特性LocalEventBus (本地事件总线)DistributedEventBus (分布式事件总线)通信边界进程内跨进程、跨服务、跨机器传输方式内存方法调用网络协议 (HTTP, TCP, AMQP 等)核心关切执行速度、内存管理网络可靠性、序列化、消息持久化、重试、幂等性依赖基础设施无RabbitMQ, Kafka, Redis, Azure Service Bus 等2.糟糕的设计：如果没有 DistributedEventBusBase ，强制让 RabbitMqDistributedEventBus 直接继承 EventBusBase，会导致什么后果？
[*]LocalEventBus 被强迫实现了它完全不需要的方法.
[*]EventBusBase 这个定义通用事件的基类，包含了分布式特有的抽象，变得臃肿且不专注。
[*]每个分布式实现（RabbitMQ, Kafka, Redis）都要在 PublishAsync 中重复编写序列化、连接管理、错误重试等大量公共代码。
[*]难以维护：任何对分布式公共逻辑的修改都需要在所有具体的实现类中进行，极易出错。
3.优秀的设计：引入 DistributedEventBusBase 抽象层
[*]EventBusBase: 定义事件总线的最基础、最通用契约总线逻辑，它不关心消息是如何被传输的，只关心如何找到处理器并执行。这部分逻辑本地和分布式是共享的。
[*]DistributedEventBusBase: 作为分布式事件总线的抽象起点，处理和实现分布式场景下的公共逻辑。
如何实现一个自定义的 Redis 分布式事件总线？
1.创建一个 RedisDistributedEventBus 类，继承自 DistributedEventBusBase
2.然后主要就是实现 PublishToMessageBrokerAsync (发布消息) 和 SubscribeAsync (订阅消息)
4.3.本地事件具体执行流程
经过上面的分析，本地事件的发送职责完全是由LocalEventBus中的PublishAsync及其重载来承担的，对消息进行业务加工完成，最终调用具体的handler是由父类的TriggerHandlersAsync方法完成的，这里澄清一下，有点绕容易误解
[*]LocalEventBus中的PublishAsync对实际的消息进行本地事件的业务加工，同理分布式事件的也有这样的业务加工步骤，例如分布式需要对消息进行各自的序列化，这个只能自己实现，你不可能写到父类中吧，至于为什么，看上面的问题疑问。
我们直接看TriggerHandlersAsync中的实现吧，它是所有的类型的事件总线都可以共用的，因为不管是本地事件还是分布式事件，最终调用执行订阅时的Handler的逻辑是一样的，这里的逻辑
说的是技术实现， 如果不好理解就这么想， 我有消息事件对象了，下一步就是要在事件订阅缓存映射列表中找到具体的事件执行Handler而找到handler 并且执行handler的逻辑这部分
所有类型的事件总线中都是共用的。
这里主要执行的逻辑
4.3.1.事件处理器触发
1.调用GetHandlerFactories() ，根据类型取出事件订阅缓存映射列表中的映射元素。
2.然后使用IEventHandlerInvoker用以 执行handler的HandleAsync方法。
我们来具体分析下这部分他是如何实现的，框架在设计时抽象了2个接口，一个是用于调用事件handler的IEventHandlerInvoker 一个是 用于具体执行的
IEventHandlerMethodExecutor，他们的调用方向如下:
可以思考一下，为什么使用接口？
首先EventBusBase中依赖IEventHandlerInvoker接口， 他在构造时就已经注入，当调用InvokeAsync()时，其实是它的实例EventHandlerInvoker来负责具体执行
在EventHandlerInvoker内部组合了一个缓存字典，它用于缓存执行器，好处是只需要匹配一次，避免每次都要去使用反射创建执行器实例。
匹配到具体的执行器之后就是具体执行了，执行的动作分别交给了LocalEventHandlerMethodExecutor和DistributedEventHandlerMethodExecutor
看一下执行器内部实现，很有意思，这里是最简单的一种方式，不过这种思路，在框架中有很多类似的案例，但是实现方式不一样，可以使用表达式树 、Emit 动态生成IL、MethodInfo.Invoke（反射）、Delegate.CreateDelegate，感兴趣的可以自定义扩展IEventHandlerMethodExecutor试试
4.3.2.事件传播机制
事件总线中的“继承传播”机制，目的是当一个泛型事件被发布时，自动将该事件也以它的父类泛型形式发布一次，从而让监听其父类类型的订阅者也能收到通知。
啥意思呢？直接看例子吧，注意必须实现IEventDataWithInheritableGenericArgument接口，然后事件对象必须是泛型的
class Entity { }
class User : Entity { }
//必须实现IEventDataWithInheritableGenericArgument接口，然后事件对象必须是泛型的
class EntityCreatedEvent<T> : IEventDataWithInheritableGenericArgument
{
public T Entity { get; }
public EntityCreatedEvent(T entity) => Entity = entity;
object[] GetConstructorArgs() => new object[] { Entity };
}
var userEvent = new EntityCreatedEvent<User>(new User());当发布 userEvent 时：
[*]监听 EntityCreatedEvent 的事件处理器能收到
[*]监听 EntityCreatedEvent 的事件处理器也能收到
5.总结
我们对ABP框架中事件总线（Event Bus）模块的设计与实现，围绕“订阅”和“发布”两大机制展开。介绍了基本用法，也逐步深入分析了其背后的设计思想、类结构、执行流程以及扩展性考虑。
核心设计思想： 事件总线作为发布-订阅模式的中介者，它的核心是解耦发布者与订阅者。ABP通过抽象（IEventBus, IEventHandler）和分层（EventBusBase -> LocalEventBus/DistributedEventBusBase）设计，提供一个通用、灵活且可扩展的实现。
关键实现机制：
订阅：维护一个 ConcurrentDictionary 结构来映射事件类型和处理工厂。通过依赖注入自动扫描注册和手动API注册两种方式填充映射字典。
发布：发布时根据事件类型从字典中找出所有对应的 IEventHandlerFactory，由 IEventHandlerInvoker 协调，通过合适的 IEventHandlerMethodExecutor 执行具体的处理逻辑。
生命周期管理：通过不同的 IEventHandlerFactory 实现（IocEventHandlerFactory, SingleInstanceHandlerFactory, TransientEventHandlerFactory）来精确控制事件处理器的生命周期。
分层设计：为何分布式事件总线需要额外的抽象层 （DistributedEventBusBase），是为了处理序列化、网络传输等特有问题，避免污染核心通用逻辑，体现了“单一职责”和“接口隔离”原则。
最后贴上整个事件总线的类图分析，画的不完善，仅供学习
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]