.Net 事件总线之Autofac解耦

事件总线是通过一个中间服务，剥离了常规事件的发布与订阅(消费)强依赖关系的一种技术实现。事件总线的基础知识可参考圣杰的博客【事件总线知多少】
<https://www.cnblogs.com/sheng-jie/p/6970091.html>

本片博客不再详细概述事件总线基础知识，核心点放置使用Autofac组件实现事件总线与事件事件处理对象的解耦，并在实际业务场景中使用事件总线解决需求。

案例需求

这里还是先面向业务来针对性的探究下事件总线在实际业务场景里的用处有多大，再来讲解后续的Autofac解耦。

在基础数据管理模块里，需要对产品类别删除后也将相应的产品类别标签、产品类别下的产品进行删除。

我们过往的做法是在产品类别删除的业务逻辑后继续编写删除产品类别标签、产品类别对应的产品，类似于下面的代码。
private void EventBusTest1(long productCategoryId) {
_logger.LogDebug($"删除产品类别{productCategoryId}.");
_logger.LogDebug("删除产品类别标签.."); _logger.LogDebug("删除产品类别与产品关系.."); }

这种做法本身可以实现我们的实际需求，但是试想如果这个时候我需要再加一个功能，针对删除产品类别后邮件通知管理员。我们发现要实现此功能得继续在之前的代码块中加入通知邮件的相关服务。如下：
private void EventBusTest1(long productCategoryId) { _logger.LogDebug($"
删除产品类别{productCategoryId}."); _logger.LogDebug("删除产品类别标签.."); _logger.LogDebug("
删除产品类别与产品关系.."); _logger.LogDebug("发送邮件通知管理员.."); }

上面的范例代码是将业务代码采用Logger打印日志模拟出来。如上代码，实际工作中也因为经常性对某一事件进行业务逻辑补充而无限扩张部分代码块，到后来还要进行大面积重构。

ok，如何让我们产品因后续某项事件处理补充而不影响原有的代码块呢？不着急，捋一捋逻辑。


我们先思考下，删除产品类别是不是一个事件，那么它是一定有事件源对象，事件源对象即是我们针对删除产品类别所需业务处理的参数。无它我们什么也做不了，事件源EventData就是这么重要，脑海里想象下某个微信公众号，事件源就是这个公众号发布的文章。那么上面的代码中，入参productCategoryId就是一个事件源(当然我们可以包装成一个对象，补充多一点信息)


那么有了EventData后还是要把业务逻辑从聚合状态分解，不然还是聚合在一起处理和原来的处理方式一样。理想情况下是删除产品类别本身是一个Handler，删除产品类别也是一个Handler，产出产品类别与产品关系及发送邮件通知也是如此。只要触发删除产品类别，则这些Handler也一并会执行。

同一个事件源但凡绑定了多个Handler，则这些Handler会根据事件源进行各自的逻辑处理。

 



上面对EventData和EventHandler进行了简单介绍，但是我们要怎么触发呢？事件源与处理程序对象如何映射呢？


答案是采取EventBus统一对事件源进行事件触发，事件源于处理程序对象映射在一个字典里。触发事件时EventBus从字典中获取事件源对应的处理对象来分发处理。

哦豁，概念终于唠叨完了，看看使用了事件总线后是如何处理的吧，上码说话。。

先定义事件源接口
/// <summary> /// 定义事件源接口，所有的事件源都要实现该接口 /// </summary> public interface
IEventData { /// <summary> /// 事件发生的时间 /// </summary> DateTime EventTime { get;
set; } /// <summary> /// 触发事件的对象 /// </summary> object EventSource { get; set;
} }
再定义事件源基类
/// <summary> /// 事件源基类：描述事件信息，用于参数传递 /// </summary> public class EventData :
IEventData { /// <summary> /// 事件发生的时间 /// </summary> public DateTime EventTime
{ get; set; } /// <summary> /// 触发事件的对象 /// </summary> public Object
EventSource { get; set; } public EventData() { EventTime = DateTime.Now; } }
　　

定义事件处理接口
/// <summary> /// 定义事件处理器公共接口，所有的事件处理都要实现该接口 /// </summary> public interface
IEventHandler: IDependency { } /// <summary> /// 泛型事件处理器接口 /// </summary> ///
<typeparam name="TEventData"></typeparam> public interface
IEventHandler<TEventData> : IEventHandler where TEventData : IEventData { ///
<summary> /// 事件处理器实现该方法来处理事件 /// </summary> /// <param
name="eventData"></param> void HandleEvent(TEventData eventData); }
　　

定义事件源与事件处理对象存储容器接口
/// <summary> /// 定义事件源与事件处理对象存储容器接口 /// </summary> public interface
IEventStore { void AddRegister<T, TH>(string keyName) where T : IEventData
where TH : IEventHandler; void AddRegister(Type eventData, string handlerName,
Type eventHandler); void RemoveRegister<T, TH>() where T : IEventData where TH
: IEventHandler; void RemoveRegister(Type eventData, Type eventHandler); bool
HasRegisterForEvent<T>() where T : IEventData; bool HasRegisterForEvent(Type
eventData); IEnumerable<Type> GetHandlersForEvent<T>() where T : IEventData;
IEnumerable<Type> GetHandlersForEvent(Type eventData); Type
GetEventTypeByName(string eventName); bool IsEmpty { get; } void Clear(); }
实现IEventStore，这里将事件处理对象与事件源映射信息存储在内存中(无需持久化)
public class InMemoryEventStore : IEventStore { /// <summary> /// 定义锁对象 ///
</summary> private static readonly object LockObj = new object(); private
readonly ConcurrentDictionary<ValueTuple<Type, string>, Type>
_eventAndHandlerMapping; public InMemoryEventStore() { _eventAndHandlerMapping
= new ConcurrentDictionary<ValueTuple<Type, string>, Type>(); } public void
AddRegister<T, TH>(string keyName) where T : IEventData where TH :
IEventHandler { AddRegister(typeof(T), keyName, typeof(TH)); } public void
AddRegister(Type eventData, string handlerName, Type eventHandler) { lock
(LockObj) { var mapperKey = new ValueTuple<Type, string>(eventData,
handlerName); //是否存在事件参数对应的事件处理对象 if (!HasRegisterForEvent(eventData)) {
_eventAndHandlerMapping.TryAdd(mapperKey, eventHandler); } else {
_eventAndHandlerMapping[mapperKey] = eventHandler; } } } public void
RemoveRegister<T, TH>() where T : IEventData where TH : IEventHandler { var
handlerToRemove = FindRegisterToRemove(typeof(T), typeof(TH));
RemoveRegister(typeof(T), handlerToRemove); } public void RemoveRegister(Type
eventData, Type eventHandler) { if (eventHandler != null) { lock (LockObj) {
//移除eventHandler var eventHandelerBind =
_eventAndHandlerMapping.FirstOrDefault(p => p.Value == eventHandler); if
(eventHandelerBind.Value != null) { Type removedHandlers;
_eventAndHandlerMapping.TryRemove(eventHandelerBind.Key, out removedHandlers);
} } } } private Type FindRegisterToRemove(Type eventData, Type eventHandler) {
if (!HasRegisterForEvent(eventData)) { return null; } return
_eventAndHandlerMapping.FirstOrDefault(p => p.Value == eventHandler).Value; }
public bool HasRegisterForEvent<T>() where T : IEventData { var mapperDto =
_eventAndHandlerMapping.FirstOrDefault(p => p.Key.Item1 == typeof(T)); return
mapperDto.Value != null ? true : false; } public bool
HasRegisterForEvent(ValueTuple<Type, string> mapperKey) { return
_eventAndHandlerMapping.ContainsKey(mapperKey); } public bool
HasRegisterForEvent(Type eventData) { return _eventAndHandlerMapping.Count(p =>
p.Key.Item1 == eventData) > 0 ? true : false; } public IEnumerable<Type>
GetHandlersForEvent<T>() where T : IEventData { return
GetHandlersForEvent(typeof(T)); } public IEnumerable<Type>
GetHandlersForEvent(Type eventData) { if (HasRegisterForEvent(eventData)) { var
items = _eventAndHandlerMapping .Where(p => p.Key.Item1 == eventData) .Select(p
=> p.Value).ToList(); return items; } return new List<Type>(); } public Type
GetEventTypeByName(string eventName) { return
_eventAndHandlerMapping.Keys.FirstOrDefault(eh => eh.Item2 == eventName).Item1;
} public bool IsEmpty => !_eventAndHandlerMapping.Keys.Any(); public void
Clear() => _eventAndHandlerMapping.Clear(); }
　　

在定义EventHandler，我们这里定义3个EventHandler，分别对应产品类别删除、产品类别标签删除、产品类别与产品关系信息删除 
public class DeleteProductCategoryEventHandler :
IEventHandler<DeleteProductCategoryEventData> { private readonly ILogger
_logger; public
DeleteProductCategoryEventHandler(ILogger<DeleteProductCategoryEventHandler>
logger) { _logger = logger; } public void
HandleEvent(DeleteProductCategoryEventData eventData) {
_logger.LogDebug($"删除产品类别{eventData.ProductCategoryId}.."); } } public class
DeleteProductCategoryTagEventHandler :
IEventHandler<DeleteProductCategoryEventData> { private readonly ILogger
_logger; public
DeleteProductCategoryTagEventHandler(ILogger<DeleteProductCategoryTagEventHandler>
logger) { _logger = logger; } public void
HandleEvent(DeleteProductCategoryEventData eventData) {
_logger.LogDebug($"删除产品类别{eventData.ProductCategoryId}标签.."); } } public class
DeleteProductCategoryRelEventHandler :
IEventHandler<DeleteProductCategoryEventData> { private readonly ILogger
_logger; public
DeleteProductCategoryRelEventHandler(ILogger<DeleteProductCategoryRelEventHandler>
logger) { _logger = logger; } public void
HandleEvent(DeleteProductCategoryEventData eventData) {
_logger.LogDebug($"删除产品类别{eventData.ProductCategoryId}与产品关系.."); } }
　　


 还有最重要的EventBus，我们使用它来同一触发Handler，EventBus会从EventStore获取到事件源映射的Handler集合，并通过DI容器实例化对象后执行事件。

EventBus
/// <summary> /// 事件总线 /// </summary> public class EventBus : IEventBus {
private readonly IEventStore _eventStore; public static EventBus Default { get;
} = new EventBus(); public EventBus() { _eventStore =
ContainerManager.Current.ResolveNamed<IEventStore>(AppConst.IN_MEMORY_EVENT_STORE);
} public void Trigger<TEventData>(TEventData eventData) where TEventData :
IEventData { //从事件映射集合里获取匹配当前EventData(事件源数据)的Handler List<Type> handlerTypes =
_eventStore.GetHandlersForEvent(eventData.GetType()).ToList(); if
(handlerTypes.Count <= 0) return; //循环执行事件处理函数 foreach (var handlerType in
handlerTypes) { var handlerInterface =
handlerType.GetInterface("IEventHandler`1");
//这里需要根据Name才能Resolve，因为注入服务时候使用了命名方式(解决同一事件参数多个事件处理类绑定问题)+ var eventHandler =
ContainerManager.Current.Container.ResolveNamed(handlerType.Name,
handlerInterface); if (eventHandler.GetType().FullName == handlerType.FullName)
{ var handler = eventHandler as IEventHandler<TEventData>;
handler?.HandleEvent(eventData); } } } public void Trigger<TEventData>(Type
eventHandlerType, TEventData eventData) where TEventData : IEventData { if
(_eventStore.HasRegisterForEvent<TEventData>()) { var handlers =
_eventStore.GetHandlersForEvent<TEventData>(); if (handlers.Any(th => th ==
eventHandlerType)) { //获取类型实现的泛型接口 var handlerInterface =
eventHandlerType.GetInterface("IEventHandler`1"); var eventHandlers =
ContainerManager.Current.Container.Resolve(handlerInterface);
//循环遍历，仅当解析的实例类型与映射字典中事件处理类型一致时，才触发事件 if (eventHandlers.GetType() ==
eventHandlerType) { var handler = eventHandlers as IEventHandler<TEventData>;
handler?.HandleEvent(eventData); } } } } public Task
TriggerAsycn<TEventData>(Type eventHandlerType, TEventData eventData) where
TEventData : IEventData { return Task.Run(() => Trigger(eventHandlerType,
eventData)); } public Task TriggerAsync<TEventData>(TEventData eventData) where
TEventData : IEventData { return Task.Run(() => Trigger(eventData)); } public
void UnRegister<TEventData>(Type handlerType) where TEventData : IEventData {
_eventStore.RemoveRegister(typeof(TEventData), handlerType); } public void
UnRegisterAll<TEventData>() where TEventData : IEventData {
//获取所有映射的EventHandler List<Type> handlerTypes =
_eventStore.GetHandlersForEvent(typeof(TEventData)).ToList(); foreach (var
handlerType in handlerTypes) { _eventStore.RemoveRegister(typeof(TEventData),
handlerType); } } }
　　


当然最重要的是如何使用DI容器注入事件处理程序以及相关依赖服务，这里要注意Autofac对同一个容器只允许Build一次或Update一次。即不允许在程序运行时动态注入服务。且
针对同一接口除非使用命名方式注入服务，否则默认一个接口映射一个服务
。Autofac这个特性极大程度限制了同一事件源多个Handler的情况，没办法我还是想了个办法，
参考AutoMapper的Profile映射方式，定义一个用于映射的类，如下：
public class EventRegisterService : IEventRegisterService { public void
RegisterEventHandler(IEventStore _eventStore, ContainerBuilder builder) {
builder.RegisterType<DeleteProductCategoryEventHandler>().Named<IEventHandler<DeleteProductCategoryEventData>>("DeleteProductCategoryEventHandler");
_eventStore.AddRegister(typeof(DeleteProductCategoryEventData),
"DeleteProductCategoryEventHandler",
typeof(DeleteProductCategoryEventHandler));
builder.RegisterType<DeleteProductCategoryTagEventHandler>().Named<IEventHandler<DeleteProductCategoryEventData>>("DeleteProductCategoryTagEventHandler");
_eventStore.AddRegister(typeof(DeleteProductCategoryEventData),
"DeleteProductCategoryTagEventHandler",
typeof(DeleteProductCategoryTagEventHandler));
builder.RegisterType<DeleteProductCategoryRelEventHandler>().Named<IEventHandler<DeleteProductCategoryEventData>>("DeleteProductCategoryRelEventHandler");
_eventStore.AddRegister(typeof(DeleteProductCategoryEventData),
"DeleteProductCategoryRelEventHandler",
typeof(DeleteProductCategoryRelEventHandler)); } }
通过上面的映射后，在程序初始化时可使用C#反射的特性执行 RegisterEventHandler实现注入同一事件源多个事件处理服务，简直完美~~

下面是Autofac注入的代码：
/// <summary> /// 通过反射注入事件处理对象 /// </summary> /// <param
name="assemblieLst"></param> /// <param name="_eventStore"></param> private
static void RegisterAllEventHandlerFromAssembly(List<Assembly> assemblieLst,
IEventStore _eventStore) { var baseType = typeof(IEventRegisterService);
foreach (var assem in assemblieLst) { assem.GetTypes().Each(p => { if
(baseType.IsAssignableFrom(p) && p.IsClass) { //反射执行注册方法 object obj =
Activator.CreateInstance(p); MethodInfo method =
obj.GetType().GetMethod(_eventBusBindMethod); method.Invoke(obj, new object[] {
_eventStore, Builder }); } }); } } /// <summary> /// 注册服务 /// </summary> ///
<param name="builder"></param> /// <param name="assemblieLst"></param> private
static IContainer RegisterDependencyService(List<Assembly> assemblieLst,
IServiceCollection services) { //1、注入依赖服务 Builder = new ContainerBuilder();
Builder.Populate(services); var baseType = typeof(IDependency); foreach (var
assem in assemblieLst) { Builder.RegisterAssemblyTypes(assem) .Where(type =>
baseType.IsAssignableFrom(type) && !type.IsAbstract) .AsImplementedInterfaces()
.SingleInstance(); } //2、注入AutoMaper配置 List<Type> loadedProfiles =
RetrieveProfiles(assemblieLst);
Builder.RegisterTypes(loadedProfiles.ToArray()); //3、注入事件总线 var eventStore =
new InMemoryEventStore(); Builder.Register(c =>
eventStore).Named<IEventStore>(AppConst.IN_MEMORY_EVENT_STORE).SingleInstance();
RegisterAllEventHandlerFromAssembly(assemblieLst, eventStore); //4、生成容器
IContainer container = Builder.Build(); RegisterAutoMapper(container,
loadedProfiles); return container; }
　　

OK，看下最终效果，将触发事件写在Action中



 

F5跑起来，Debug目录下出来了相应的日志:



内容如下:



 

 

总结


事件总线是个好东西，使用它可以将业务分散处理，唯一麻烦点在于需要保证业务处理的原子性问题。以前我们直接一个事务套起来就完事，现在可能要采取类似微服务中的“最终一致性”方案，所以合适的业务采用合适的技术。

时间总线也是领域驱动设计里重要的知识点，掌握它很有必要，微服务中也是采取它配合消息队列实现跨服务处理领域事件。这里使用Autofac也是因为自己的架构本身就是使用它做默认的DI容器，在与事件总线融合中也发现了Autofac的局限性，那就是运行时注册，如果能运行时注册服务的话真的是不要太爽了。

最后如果有长沙的小伙伴可以加入.NET长沙社区，大家共同探讨.NET Core、Redis、Docker等流行技术，共同进步！
微信一群已加满，大家可加入微信二群，也可以扫下面支付码请我喝杯奶茶 o(∩_∩)o 哈哈~~

            

 

 

 

• « 上一篇：sparkstreaming消费kafka的数据手动维护offest实现exactly-once语义
• » 下一篇：Etcd在kubernetes集群中的作用