经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » 编程经验 » 查看文章
将Abp默认事件总线改造为分布式事件总线
来源：cnblogs　　作者：林晓lx　　时间：2023/12/21 9:02:59　　对本文有异议

@

目录

原理

本地事件总线是通过Ioc容器来实现的。

IEventBus接口定义了事件总线的基本功能，如注册事件、取消注册事件、触发事件等。

Abp.Events.Bus.EventBus是本地事件总线的实现类，其中私有成员ConcurrentDictionary<Type, List<IEventHandlerFactory>> _handlerFactories是事件订阅表。通过维护事件订阅表来实现事件处理器的注册和取消注册。当对应类型的事件触发时，通过订阅表查找所有事件处理器，通过Ioc容器来获取处理器实例，然后通过反射来调用事件处理器的"HandleEvent"方法。

创建分布式事件总线

首先，我们需要一个分布式事件总线中间件，用来将事件从本地事件总线转发到分布式事件总线。常用的中间件有RabbitMQ、Kafka、Redis等。

开源社区已经有实现好的库，本项目参考了 wuyi6216/Abp.RemoteEventBus

这里已经定义好了一个分布式事件总线接口

  2. public interface IDistributedEventBus : IDisposable
  3. {
  4.     void MessageHandle(string topic, string message);
  6.     void Publish(IDistributedEventData eventData);
  8.     void Subscribe(string topic);
  10.     void Unsubscribe(string topic);
  12.     void UnsubscribeAll();
  13. }

为了兼容本地事件总线，我们需要定义一个分布式事件总线接口，继承自IEventBus接口。

  2. public interface IMultipleEventBus : IDistributedEventBus, IEventBus
  3. {
  5. }

实现自动订阅和事件转发

当注册本地事件时，将订阅分布式事件，事件Topic为类型的字符串表现形式

  1. public IDisposable Register(Type eventType, IEventHandlerFactory factory)
  2. {
  3.     GetOrCreateHandlerFactories(eventType);
  4.     List<IEventHandlerFactory> currentLists;
  5.     if (_handlerFactories.TryGetValue(eventType, out currentLists))
  6.     {
  7.         lock (currentLists)
  8.         {
  9.             if (currentLists.Count == 0)
  10.             {
  11.                 //Register to distributed event
  12.                 this.Subscribe(eventType.ToString());
  13.             }
  14.             currentLists.Add(factory);
  15.         }
  16.     }
  17.     return new FactoryUnregistrar(this, eventType, factory);
  18. }

创建TriggerRemote，此方法用于将本地事件参数打包成为分布式事件消息payload，并发布该消息

  1. public void TriggerRemote(Type eventType, object eventSource, IEventData eventData)
  2. {
  3.     var exceptions = new List<Exception>();
  4.     eventData.EventSource = eventSource;
  5.     try
  6.     {
  7.         var payloadDictionary = new Dictionary<string, object>
  8.                 {
  9.                     { PayloadKey, eventData }
  10.                 };
  11.         var distributedeventData = new DistributedEventData(eventType.ToString(), payloadDictionary);
  12.         Publish(distributedeventData);
  13.     }
  15.     catch (Exception ex)
  16.     {
  17.         exceptions.Add(ex);
  18.     }
  19.     if (exceptions.Any())
  20.     {
  21.         if (exceptions.Count == 1)
  22.         {
  23.             exceptions[0].ReThrow();
  24.         }
  26.         throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions);
  27.     }
  28. }

当触发本地事件时，将消息转发至分布式事件总线。
在Trigger方法中调用TriggerRemote，事件状态回调和事件异常回调将不会被转发。

  1. if (!(typeof(DistributedEventBusEvent) == eventType
  2.    || typeof(DistributedEventBusEvent).IsAssignableFrom(eventType)
  3.    || typeof(DistributedEventMessageHandleExceptionData) == eventType
  4.    || typeof(DistributedEventHandleExceptionData) == eventType
  5.     ))
  6. {
  7.     if (typeof(DistributedEventArgs) != eventType)
  8.     {
  9.         TriggerRemote(eventType, eventSource, eventData);
  11.     }
  12. }

在消费端接收到分布式事件消息时，从Topic中解析类型，转发给本地事件。若此类型在本地事件注册过，则将消息反序列化为本地事件参数，然后触发本地事件。
本地事件处理器将触发最终的处理方法。

  2. public virtual void MessageHandle(string topic, string message)
  3. {
  4.     Logger.Debug($"Receive message on topic {topic}");
  5.     try
  6.     {
  7.         var eventData = _remoteEventSerializer.Deserialize<DistributedEventData>(message);
  8.         var eventArgs = new DistributedEventArgs(eventData, topic, message);
  9.         Trigger(this, new DistributedEventBusHandlingEvent(eventArgs));
  11.         if (!string.IsNullOrEmpty(eventData.Type))
  12.         {
  13.             string pattern = @"(.*?)\[(.*?)\]";
  14.             Match match = Regex.Match(eventData.Type, pattern);
  15.             if (match.Success)
  16.             {
  18.                 var type = match.Groups[1].Value;
  19.                 var type2 = match.Groups[2].Value;
  21.                 var localTriggerType = typeFinder.Find(c => c.FullName == type).FirstOrDefault();
  22.                 var genericType = typeFinder.Find(c => c.FullName == type2).FirstOrDefault();
  24.                 if (localTriggerType != null && genericType != null)
  25.                 {
  27.                     if (localTriggerType.GetTypeInfo().IsGenericType
  28.                         && localTriggerType.GetGenericArguments().Length == 1
  29.                         && !genericType.IsAbstract && !genericType.IsInterface
  30.                         )
  31.                     {
  32.                         var localTriggerGenericType = localTriggerType.GetGenericTypeDefinition().MakeGenericType(genericType);
  35.                         if (eventData.Data.TryGetValue(PayloadKey, out var payload))
  36.                         {
  37.                             var payloadObject = (payload as JObject).ToObject(localTriggerGenericType);
  38.                             Trigger(localTriggerGenericType, this, (IEventData)payloadObject);
  40.                         }
  41.                     }
  42.                 }
  45.             }
  46.             else
  47.             {
  48.                 var localTriggerType = typeFinder.Find(c => c.FullName == eventData.Type).FirstOrDefault();
  49.                 if (localTriggerType != null && !localTriggerType.IsAbstract && !localTriggerType.IsInterface)
  50.                 {
  51.                     if (eventData.Data.TryGetValue(PayloadKey, out var payload))
  52.                     {
  53.                         var payloadObject = (payload as JObject).ToObject(localTriggerType);
  54.                         Trigger(localTriggerType, this, (IEventData)payloadObject);
  56.                     }
  58.                 }
  59.             }
  60.             Trigger(this, new DistributedEventBusHandledEvent(eventArgs));
  62.         }
  63.     }
  64.     catch (Exception ex)
  65.     {
  66.         Logger.Error("Consume remote message exception", ex);
  67.         Trigger(this, new DistributedEventMessageHandleExceptionData(ex, topic, topic));
  68.     }
  69. }

使用

DistributedEventBus有不同的实现方式，这里以Redis为例

启动Redis服务

下载Redis并启动服务，使用默认端口6379

配置

生产者和消费者端都需要配置分布式事件总线

首先引用Abp.DistributedEventBus.Redis，并配置Abp模块依赖

[DependsOn(typeof(AbpDistributedEventBusRedisModule))]

在PreInitialize方法中配置Redis连接信息

  1.  Configuration.Modules.DistributedEventBus().UseRedis().Configure(setting =>
  2.  {
  3.      setting.Server = "127.0.0.1:6379";
  4.  });

用MultipleEventBus替换Abp默认事件总线

  1.  //todo: 事件总线
  2.  Configuration.ReplaceService(
  3.   typeof(IEventBus),
  4.   () => IocManager.IocContainer.Register(
  5.       Component.For<IEventBus>().ImplementedBy<MultipleEventBus>()
  6.   ));

传递Abp默认事件

我们知道在使用仓储时，Abp会自动触发一些事件，如创建、更新、删除等。我们来测试这些事件是否能通过分布式事件总线来传递。

定义一个实体类，用于传递实体的增删改事件。

  2. public class Person : FullAuditedEntity<int>
  3. {
  5.     public string Name { get; set; }
  6.     public int Age { get; set; }
  7.     public string PhoneNumber { get; set; }
  9. }

在消费者端，定义一个事件处理器，用于处理实体的增删改事件。

  2. public class RemoteEntityChangedEventHandler :
  3.     IEventHandler<EntityUpdatedEventData<Person>>,
  4.     IEventHandler<EntityCreatedEventData<Person>>,
  5.     IEventHandler<EntityDeletedEventData<Person>>,
  6.     ITransientDependency
  7. {
  9.     void IEventHandler<EntityUpdatedEventData<Person>>.HandleEvent(EntityUpdatedEventData<Person> eventData)
  10.     {
  11.         var person = eventData.Entity;
  12.         Console.WriteLine($"Remote Entity Updated - Name:{person.Name}, Age:{person.Age}, PhoneNumber:{person.PhoneNumber}");
  13.     }
  15.     void IEventHandler<EntityCreatedEventData<Person>>.HandleEvent(EntityCreatedEventData<Person> eventData)
  16.     {
  17.         var person = eventData.Entity;
  18.         Console.WriteLine($"Remote Entity Created - Name:{person.Name}, Age:{person.Age}, PhoneNumber:{person.PhoneNumber}");
  20.     }
  22.     void IEventHandler<EntityDeletedEventData<Person>>.HandleEvent(EntityDeletedEventData<Person> eventData)
  23.     {
  24.         var person = eventData.Entity;
  25.         Console.WriteLine($"Remote Entity Deleted - Name:{person.Name}, Age:{person.Age}, PhoneNumber:{person.PhoneNumber}");
  27.     }
  28. }

在生产者端，用IRepository对实体进行增删改操作。

  2. var person = new Person()
  3. {
  5.     Name = "John",
  6.     Age = 36,
  7.     PhoneNumber = "18588888888"
  9. };
  11. personRepository.Insert(person);
  13. var person2 = new Person()
  14. {
  16.     Name = "John2",
  17.     Age = 36,
  18.     PhoneNumber = "18588888889"
  20. };
  21. personRepository.Insert(person2);
  23. var persons = personRepository.GetAllList();
  24. foreach (var p in persons)
  25. {
  26.     p.Age += 1;
  27.     personRepository.Update(p);
  28.     Console.WriteLine($"Entity Updated - Name:{p.Name}, Age:{p.Age}, PhoneNumber:{p.PhoneNumber}");
  30. }
  31. foreach (var p in persons)
  32. {
  33.     personRepository.Delete(p);
  34.     Console.WriteLine($"Entity Deleted - Name:{p.Name}, Age:{p.Age}, PhoneNumber:{p.PhoneNumber}");
  36. }

运行程序（同时运行消费者端和生产者端），可以看到消费者端打印出了实体的增删改事件。

在这里插入图片描述

注意：

分布式事件总线在两个独立系统间传递事件，所以需要定义一个共同的类型对象，用于事件参数的传递。
因此消费者端需要引用生产者端的模块，以便获取共同的类型对象。

  1. public override Assembly[] GetAdditionalAssemblies()
  2. {
  3.     var clientModuleAssembly = typeof(Person).GetAssembly();
  4.     return [clientModuleAssembly];
  5. }

传递自定义事件

定义NotificationEventData，用于传递自定义事件。

  2. public class NotificationEventData : EventData
  3. {
  4.     public int Id { get; set; }
  5.     
  6.     public string Title { get; set; }
  8.     public string Message { get; set; }
  10.     public bool IsRead { get; set; }
  11. }

在消费者端，定义一个事件处理器，用于处理自定义事件。

  1. public class NotificationEventHandler :
  2.     IEventHandler<NotificationEventData>,      
  3.     ITransientDependency
  4. {
  5.     
  6.     void IEventHandler<NotificationEventData>.HandleEvent(NotificationEventData eventData)
  7.     {
  8.         Console.WriteLine($"Id: {eventData.Id}");
  9.         Console.WriteLine($"Title: {eventData.Title}");
  10.         Console.WriteLine($"Message: {eventData.Message}");
  11.         Console.WriteLine($"IsRead: {eventData.IsRead}");
  13.     }
  14. }

在生产者端，触发自定义事件。

  1. var eventBus = IocManager.Instance.Resolve<IEventBus>();
  4. eventBus.Trigger<NotificationEventData>(new NotificationEventData()
  5. {
  6.     Title = "Hi",
  7.     Message = "Customized definition event test!",
  8.     Id = 100,
  9.     IsRead = true,
  10. });

运行程序（同时运行消费者端和生产者端），可以看到消费者端打印出了自定义事件。

在这里插入图片描述

项目地址

Github:DistributedEventBus

原文链接：https://www.cnblogs.com/jevonsflash/p/17917031.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

编程经验热门文章

  • Prometheus监控之SNMP Exporter介绍和数据展现
  • Intel 82599网卡异常挂死原因
  • 腾讯防水墙的分析识别与破解
  • 桌面应用自动化WinAppDriver入门
  • 文档在线预览（三）使用js前端实现word、excel、pdf、ppt 在线预览
  • 「教程」如何下载AcFun网站上的视频、A站视频下载工具
  • 基于Hi3559AV100的SVP(NNIE)开发整体流程
  • GBDT(梯度提升树)scikit-klearn中的参数说明及简汇
  • antd之RangePicker设置默认值方式
  • 决胜经典算法之插入排序

    • 编程经验推荐文章

  • wifi基础(一)：无线电波与WIFI信号干扰、衰减
  • 甲方扔给两个存在包名与类名均相同的Jar包，要在工程中同时使用怎么办？
  • 《花100块做个摸鱼小网站! 》第三篇—热搜表结构设计和热搜数据存储
  • Fluent Editor：一个基于 Quill 2.0 的富文本编辑器，功能强大、开箱即用！
  • 方法的三种调用形式
  • 《花100块做个摸鱼小网站! 》第二篇—后端应用搭建和完成第一个爬虫
  • 《熬夜整理》保姆级系列教程-玩转Wireshark抓包神器教程(4)-再识Wireshark - 北京-宏哥
  • 《软件性能测试分析与调优实践之路》(第2版) 读书笔记（二）总体介绍（下）-真正从性能分析与调优来看性能测试
  • 架构知识点（三）
  • 《花100块做个摸鱼小网站! 》第一篇—买云服务器和初始化环境

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号