将Abp默认事件总线改造为分布式事件总线

2023-12-21 06:44
文章标签 事件 分布式 总线 默认 改造 abp

本文主要是介绍将Abp默认事件总线改造为分布式事件总线，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • 原理
    • 创建分布式事件总线
    • 实现自动订阅和事件转发
  • 使用
    • 启动Redis服务
    • 配置
    • 传递Abp默认事件
    • 传递自定义事件
  • 项目地址

原理

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

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

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

创建分布式事件总线

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

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

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


public interface IDistributedEventBus : IDisposable
{void MessageHandle(string topic, string message);void Publish(IDistributedEventData eventData);void Subscribe(string topic);void Unsubscribe(string topic);void UnsubscribeAll();
}

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


public interface IMultipleEventBus : IDistributedEventBus, IEventBus
{}

实现自动订阅和事件转发

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

public IDisposable Register(Type eventType, IEventHandlerFactory factory)
{GetOrCreateHandlerFactories(eventType);List<IEventHandlerFactory> currentLists;if (_handlerFactories.TryGetValue(eventType, out currentLists)){lock (currentLists){if (currentLists.Count == 0){//Register to distributed eventthis.Subscribe(eventType.ToString());}currentLists.Add(factory);}}return new FactoryUnregistrar(this, eventType, factory);
}

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

public void TriggerRemote(Type eventType, object eventSource, IEventData eventData)
{var exceptions = new List<Exception>();eventData.EventSource = eventSource;try{var payloadDictionary = new Dictionary<string, object>{{ PayloadKey, eventData }};var distributedeventData = new DistributedEventData(eventType.ToString(), payloadDictionary);Publish(distributedeventData);}catch (Exception ex){exceptions.Add(ex);}if (exceptions.Any()){if (exceptions.Count == 1){exceptions[0].ReThrow();}throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions);}
}

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

if (!(typeof(DistributedEventBusEvent) == eventType|| typeof(DistributedEventBusEvent).IsAssignableFrom(eventType)|| typeof(DistributedEventMessageHandleExceptionData) == eventType|| typeof(DistributedEventHandleExceptionData) == eventType))
{if (typeof(DistributedEventArgs) != eventType){TriggerRemote(eventType, eventSource, eventData);}
}

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


public virtual void MessageHandle(string topic, string message)
{Logger.Debug($"Receive message on topic {topic}");try{var eventData = _remoteEventSerializer.Deserialize<DistributedEventData>(message);var eventArgs = new DistributedEventArgs(eventData, topic, message);Trigger(this, new DistributedEventBusHandlingEvent(eventArgs));if (!string.IsNullOrEmpty(eventData.Type)){string pattern = @"(.*?)\[(.*?)\]";Match match = Regex.Match(eventData.Type, pattern);if (match.Success){var type = match.Groups[1].Value;var type2 = match.Groups[2].Value;var localTriggerType = typeFinder.Find(c => c.FullName == type).FirstOrDefault();var genericType = typeFinder.Find(c => c.FullName == type2).FirstOrDefault();if (localTriggerType != null && genericType != null){if (localTriggerType.GetTypeInfo().IsGenericType&& localTriggerType.GetGenericArguments().Length == 1&& !genericType.IsAbstract && !genericType.IsInterface){var localTriggerGenericType = localTriggerType.GetGenericTypeDefinition().MakeGenericType(genericType);if (eventData.Data.TryGetValue(PayloadKey, out var payload)){var payloadObject = (payload as JObject).ToObject(localTriggerGenericType);Trigger(localTriggerGenericType, this, (IEventData)payloadObject);}}}}else{var localTriggerType = typeFinder.Find(c => c.FullName == eventData.Type).FirstOrDefault();if (localTriggerType != null && !localTriggerType.IsAbstract && !localTriggerType.IsInterface){if (eventData.Data.TryGetValue(PayloadKey, out var payload)){var payloadObject = (payload as JObject).ToObject(localTriggerType);Trigger(localTriggerType, this, (IEventData)payloadObject);}}}Trigger(this, new DistributedEventBusHandledEvent(eventArgs));}}catch (Exception ex){Logger.Error("Consume remote message exception", ex);Trigger(this, new DistributedEventMessageHandleExceptionData(ex, topic, topic));}
}

使用

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

启动Redis服务

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

配置

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

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

[DependsOn(typeof(AbpDistributedEventBusRedisModule))]

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

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

用MultipleEventBus替换Abp默认事件总线

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

传递Abp默认事件

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

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


public class Person : FullAuditedEntity<int>
{public string Name { get; set; }public int Age { get; set; }public string PhoneNumber { get; set; }}

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


public class RemoteEntityChangedEventHandler :IEventHandler<EntityUpdatedEventData<Person>>,IEventHandler<EntityCreatedEventData<Person>>,IEventHandler<EntityDeletedEventData<Person>>,ITransientDependency
{void IEventHandler<EntityUpdatedEventData<Person>>.HandleEvent(EntityUpdatedEventData<Person> eventData){var person = eventData.Entity;Console.WriteLine($"Remote Entity Updated - Name:{person.Name}, Age:{person.Age}, PhoneNumber:{person.PhoneNumber}");}void IEventHandler<EntityCreatedEventData<Person>>.HandleEvent(EntityCreatedEventData<Person> eventData){var person = eventData.Entity;Console.WriteLine($"Remote Entity Created - Name:{person.Name}, Age:{person.Age}, PhoneNumber:{person.PhoneNumber}");}void IEventHandler<EntityDeletedEventData<Person>>.HandleEvent(EntityDeletedEventData<Person> eventData){var person = eventData.Entity;Console.WriteLine($"Remote Entity Deleted - Name:{person.Name}, Age:{person.Age}, PhoneNumber:{person.PhoneNumber}");}
}

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


var person = new Person()
{Name = "John",Age = 36,PhoneNumber = "18588888888"};personRepository.Insert(person);var person2 = new Person()
{Name = "John2",Age = 36,PhoneNumber = "18588888889"};
personRepository.Insert(person2);var persons = personRepository.GetAllList();
foreach (var p in persons)
{p.Age += 1;personRepository.Update(p);Console.WriteLine($"Entity Updated - Name:{p.Name}, Age:{p.Age}, PhoneNumber:{p.PhoneNumber}");}
foreach (var p in persons)
{personRepository.Delete(p);Console.WriteLine($"Entity Deleted - Name:{p.Name}, Age:{p.Age}, PhoneNumber:{p.PhoneNumber}");}

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

在这里插入图片描述

注意：

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

public override Assembly[] GetAdditionalAssemblies()
{var clientModuleAssembly = typeof(Person).GetAssembly();return [clientModuleAssembly];
}

传递自定义事件

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


public class NotificationEventData : EventData
{public int Id { get; set; }public string Title { get; set; }public string Message { get; set; }public bool IsRead { get; set; }
}

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

public class NotificationEventHandler :IEventHandler<NotificationEventData>,      ITransientDependency
{void IEventHandler<NotificationEventData>.HandleEvent(NotificationEventData eventData){Console.WriteLine($"Id: {eventData.Id}");Console.WriteLine($"Title: {eventData.Title}");Console.WriteLine($"Message: {eventData.Message}");Console.WriteLine($"IsRead: {eventData.IsRead}");}
}

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

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

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

在这里插入图片描述

项目地址

Github:DistributedEventBus

这篇关于将Abp默认事件总线改造为分布式事件总线的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/518985。

相关文章

Java访问修饰符public、private、protected及默认访问权限详解

Java访问修饰符public、private、protected及默认访问权限详解

《Java访问修饰符public、private、protected及默认访问权限详解》：本文主要介绍Java访问修饰符public、private、protected及默认访问权限的相关资料,每... 目录前言1. public 访问修饰符特点：示例：适用场景：2. private 访问修饰符特点：示例：
阅读更多...
java如何分布式锁实现和选型

java如何分布式锁实现和选型

《java如何分布式锁实现和选型》文章介绍了分布式锁的重要性以及在分布式系统中常见的问题和需求,它详细阐述了如何使用分布式锁来确保数据的一致性和系统的高可用性,文章还提供了基于数据库、Redis和Zo... 目录引言：分布式锁的重要性与分布式系统中的常见问题和需求分布式锁的重要性分布式系统中常见的问题和需求
阅读更多...
Golang使用etcd构建分布式锁的示例分享

Golang使用etcd构建分布式锁的示例分享

《Golang使用etcd构建分布式锁的示例分享》在本教程中,我们将学习如何使用Go和etcd构建分布式锁系统,分布式锁系统对于管理对分布式系统中共享资源的并发访问至关重要,它有助于维护一致性,防止竞... 目录引言环境准备新建Go项目实现加锁和解锁功能测试分布式锁重构实现失败重试总结引言我们将使用Go作
阅读更多...
更改docker默认数据目录的方法步骤

更改docker默认数据目录的方法步骤

《更改docker默认数据目录的方法步骤》本文主要介绍了更改docker默认数据目录的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1.查看docker是否存在并停止该服务2.挂载镜像并安装rsync便于备份3.取消挂载备份和迁
阅读更多...
Redis分布式锁使用及说明

Redis分布式锁使用及说明

《Redis分布式锁使用及说明》本文总结了Redis和Zookeeper在高可用性和高一致性场景下的应用,并详细介绍了Redis的分布式锁实现方式,包括使用Lua脚本和续期机制,最后,提到了RedLo... 目录Redis分布式锁加锁方式怎么会解错锁？举个小案例吧解锁方式续期总结Redis分布式锁如果追求
阅读更多...
Python中的异步:async 和 await以及操作中的事件循环、回调和异常

Python中的异步:async 和 await以及操作中的事件循环、回调和异常

《Python中的异步:async和await以及操作中的事件循环、回调和异常》在现代编程中,异步操作在处理I/O密集型任务时,可以显著提高程序的性能和响应速度,Python提供了asyn... 目录引言什么是异步操作？python 中的异步编程基础async 和 await 关键字asyncio 模块理论
阅读更多...
禁止平板,iPad长按弹出默认菜单事件

禁止平板,iPad长按弹出默认菜单事件

通过监控按下抬起时间差来禁止弹出事件，把以下代码写在要禁止的页面的页面加载事件里面即可     var date;document.addEventListener('touchstart', event => {date = new Date().getTime();});document.addEventListener('touchend', event => {if (new
阅读更多...
Android实现任意版本设置默认的锁屏壁纸和桌面壁纸（两张壁纸可不一致）

Android实现任意版本设置默认的锁屏壁纸和桌面壁纸（两张壁纸可不一致）

客户有些需求需要设置默认壁纸和锁屏壁纸  在默认情况下 这两个壁纸是相同的  如果需要默认的锁屏壁纸和桌面壁纸不一样 需要额外修改 Android13实现 替换默认桌面壁纸： 将图片文件替换frameworks/base/core/res/res/drawable-nodpi/default_wallpaper.*  （注意不能是bmp格式） 替换默认锁屏壁纸： 将图片资源放入vendo
阅读更多...
集中式版本控制与分布式版本控制——Git 学习笔记01

集中式版本控制与分布式版本控制——Git 学习笔记01

什么是版本控制 如果你用 Microsoft Word 写过东西，那你八成会有这样的经历： 想删除一段文字，又怕将来这段文字有用，怎么办呢？有一个办法，先把当前文件“另存为”一个文件，然后继续改，改到某个程度，再“另存为”一个文件。就这样改着、存着……最后你的 Word 文档变成了这样： 过了几天，你想找回被删除的文字，但是已经记不清保存在哪个文件了，只能挨个去找。真麻烦，眼睛都花了。看
阅读更多...
开源分布式数据库中间件

开源分布式数据库中间件

转自：https://www.csdn.net/article/2015-07-16/2825228 MyCat：开源分布式数据库中间件 为什么需要MyCat？ 虽然云计算时代，传统数据库存在着先天性的弊端，但是NoSQL数据库又无法将其替代。如果传统数据易于扩展，可切分，就可以避免单机（单库）的性能缺陷。 MyCat的目标就是：低成本地将现有的单机数据库和应用平滑迁移到“云”端
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2