动手造轮子:基于 Redis 实现 EventBus
动手造轮子:基于 Redis 实现 EventBus
Intro
上次我们造了一个简单的基于内存的 EventBus
,但是如果要跨系统的话就不合适了,所以有了这篇基于 Redis
的 EventBus
探索。本文的实现是基于 StackExchange.Redis
来实现。
RedisEventStore
实现
既然要实现跨系统的 EventBus
再使用基于内存的 EventStore
自然不行,因此这里基于 Redis
设计了一个 EventStoreInRedis
,基于 redis 的 Hash 来实现,以 Event
的 EventKey 作为 fieldName,以 Event
对应的 EventHandler
作为 Value。
EventStoreInRedis
实现:
public class EventStoreInRedis : IEventStore
{ protected readonly string EventsCacheKey; protected readonly ILogger Logger; private readonly IRedisWrapper Wrapper; public EventStoreInRedis(ILogger<EventStoreInRedis> logger) { Logger = logger; Wrapper = new RedisWrapper(RedisConstants.EventStorePrefix); EventsCacheKey = RedisManager.RedisConfiguration.EventStoreCacheKey; } public bool AddSubscription<TEvent, TEventHandler>() where TEvent : IEventBase where TEventHandler : IEventHandler<TEvent> { var eventKey = GetEventKey<TEvent>(); var handlerType = typeof(TEventHandler); if (Wrapper.Database.HashExists(EventsCacheKey, eventKey)) { var handlers = Wrapper.Unwrap<HashSet<Type>>(Wrapper.Database.HashGet(EventsCacheKey, eventKey)); if (handlers.Contains(handlerType)) { return false; } handlers.Add(handlerType); Wrapper.Database.HashSet(EventsCacheKey, eventKey, Wrapper.Wrap(handlers)); return true; } else { return Wrapper.Database.HashSet(EventsCacheKey, eventKey, Wrapper.Wrap(new HashSet<Type> { handlerType }), StackExchange.Redis.When.NotExists); } } public bool Clear() { return Wrapper.Database.KeyDelete(EventsCacheKey); } public ICollection<Type> GetEventHandlerTypes<TEvent>() where TEvent : IEventBase { var eventKey = GetEventKey<TEvent>(); return Wrapper.Unwrap<HashSet<Type>>(Wrapper.Database.HashGet(EventsCacheKey, eventKey)); } public string GetEventKey<TEvent>() { return typeof(TEvent).FullName; } public bool HasSubscriptionsForEvent<TEvent>() where TEvent : IEventBase { var eventKey = GetEventKey<TEvent>(); return Wrapper.Database.HashExists(EventsCacheKey, eventKey); } public bool RemoveSubscription<TEvent, TEventHandler>() where TEvent : IEventBase where TEventHandler : IEventHandler<TEvent> { var eventKey = GetEventKey<TEvent>(); var handlerType = typeof(TEventHandler); if (!Wrapper.Database.HashExists(EventsCacheKey, eventKey)) { return false; } var handlers = Wrapper.Unwrap<HashSet<Type>>(Wrapper.Database.HashGet(EventsCacheKey, eventKey)); if (!handlers.Contains(handlerType)) { return false; } handlers.Remove(handlerType); Wrapper.Database.HashSet(EventsCacheKey, eventKey, Wrapper.Wrap(handlers)); return true; }
}
RedisWrapper
及更具体的代码可以参考我的 Redis 的扩展的实现 https://github.com/WeihanLi/WeihanLi.Redis
RedisEventBus
实现
RedisEventBus 是基于 Redis 的 PUB/SUB 实现的,实现的感觉还有一些小问题,我想确保每个客户端注册的时候每个 EventHandler
即使多次注册也只注册一次,但是还没找到一个好的实现,如果你有什么想法欢迎指出,和我一起交流。具体的实现细节如下:
public class RedisEventBus : IEventBus
{ private readonly IEventStore _eventStore; private readonly ISubscriber _subscriber; private readonly IServiceProvider _serviceProvider; public RedisEventBus(IEventStore eventStore, IConnectionMultiplexer connectionMultiplexer, IServiceProvider serviceProvider) { _eventStore = eventStore; _serviceProvider = serviceProvider; _subscriber = connectionMultiplexer.GetSubscriber(); } private string GetChannelPrefix<TEvent>() where TEvent : IEventBase { var eventKey = _eventStore.GetEventKey<TEvent>(); var channelPrefix = $"{RedisManager.RedisConfiguration.EventBusChannelPrefix}{RedisManager.RedisConfiguration.KeySeparator}{eventKey}{RedisManager.RedisConfiguration.KeySeparator}"; return channelPrefix; } private string GetChannelName<TEvent, TEventHandler>() where TEvent : IEventBase where TEventHandler : IEventHandler<TEvent> => GetChannelName<TEvent>(typeof(TEventHandler)); private string GetChannelName<TEvent>(Type eventHandlerType) where TEvent : IEventBase { var channelPrefix = GetChannelPrefix<TEvent>(); var channelName = $"{channelPrefix}{eventHandlerType.FullName}"; return channelName; } public bool Publish<TEvent>(TEvent @event) where TEvent : IEventBase { if (!_eventStore.HasSubscriptionsForEvent<TEvent>()) { return false; } var eventData = @event.ToJson(); var handlerTypes = _eventStore.GetEventHandlerTypes<TEvent>(); foreach (var handlerType in handlerTypes) { var handlerChannelName = GetChannelName<TEvent>(handlerType); _subscriber.Publish(handlerChannelName, eventData); } return true; } public bool Subscribe<TEvent, TEventHandler>() where TEvent : IEventBase where TEventHandler : IEventHandler<TEvent> { _eventStore.AddSubscription<TEvent, TEventHandler>(); var channelName = GetChannelName<TEvent, TEventHandler>(); TODO: if current client subscribed the channel //if (true) //{ _subscriber.Subscribe(channelName, async (channel, eventMessage) => { var eventData = eventMessage.ToString().JsonToType<TEvent>(); var handler = _serviceProvider.GetServiceOrCreateInstance<TEventHandler>(); if (null != handler) { await handler.Handle(eventData).ConfigureAwait(false); } }); return true; //} //return false; } public bool Unsubscribe<TEvent, TEventHandler>() where TEvent : IEventBase where TEventHandler : IEventHandler<TEvent> { _eventStore.RemoveSubscription<TEvent, TEventHandler>(); var channelName = GetChannelName<TEvent, TEventHandler>(); TODO: if current client subscribed the channel //if (true) //{ _subscriber.Unsubscribe(channelName); return true; //} //return false; }
}
使用示例:
使用起来大体上和上一篇使用一致,只是在初始化注入服务的时候,我们需要把 IEventBus
和 IEventStore
替换为对应 Redis 的实现即可。
1. 注册服务
services.AddSingleton<IEventBus, RedisEventBus>();
services.AddSingleton<IEventStore, EventStoreInRedis>();
2. 注册 EventHandler
services.AddSingleton<NoticeViewEventHandler>();
3. 订阅事件
eventBus.Subscribe<NoticeViewEvent, NoticeViewEventHandler>();
4. 发布事件
[HttpGet("{path}")]
public async Task<IActionResult> GetByPath(string path, CancellationToken cancellationToken, [FromServices]IEventBus eventBus)
{ var notice = await _repository.FetchAsync(n => n.NoticeCustomPath == path, cancellationToken); if (notice == null) { return NotFound(); } eventBus.Publish(new NoticeViewEvent { NoticeId = notice.NoticeId }); return Ok(notice);
}
Memo
如果要实现基于消息队列的事件处理,需要注意,消息可能会重复,可能会需要在事件处理中注意一下业务的幂等性或者对消息对一个去重处理。
我在使用 Redis 的事件处理中使用了一个基于 Redis 原子递增的特性设计的一个防火墙,从而实现一段时间内某一个消息id只会被处理一次,实现源码:https://github.com/WeihanLi/ActivityReservation/blob/dev/ActivityReservation.Helper/Events/NoticeViewEvent.cs
public class NoticeViewEvent : EventBase
{ public Guid NoticeId { get; set; } // UserId // IP // ...
}
public class NoticeViewEventHandler : IEventHandler<NoticeViewEvent>
{ public async Task Handle(NoticeViewEvent @event) { var firewallClient = RedisManager.GetFirewallClient($"{nameof(NoticeViewEventHandler)}_{@event.EventId}", TimeSpan.FromMinutes(5)); if (await firewallClient.HitAsync()) { await DependencyResolver.Current.TryInvokeServiceAsync<ReservationDbContext>(async dbContext => { //var notice = await dbContext.Notices.FindAsync(@event.NoticeId); //notice.NoticeVisitCount += 1; //await dbContext.SaveChangesAsync(); var conn = dbContext.Database.GetDbConnection(); await conn.ExecuteAsync($@"UPDATE tabNotice SET NoticeVisitCount = NoticeVisitCount +1 WHERE NoticeId = @NoticeId", new { @event.NoticeId }); }); } }
}
Reference
https://github.com/WeihanLi/ActivityReservation
https://github.com/WeihanLi/WeihanLi.Redis
https://redis.io/topics/pubsub
动手造轮子:基于 Redis 实现 EventBus相关推荐
- 动手造轮子:实现一个简单的基于 Console 的日志输出
动手造轮子:实现一个简单的基于 Console 的日志输出 Intro 之前结合了微软的 Logging 框架和 Serilog 写了一个简单的日志框架,但是之前的用法都是基于 log4net.ser ...
- 动手造轮子:实现一个简单的 EventBus
动手造轮子:实现一个简单的 EventBus Intro EventBus 是一种事件发布订阅模式,通过 EventBus 我们可以很方便的实现解耦,将事件的发起和事件的处理的很好的分隔开来,很好的实 ...
- 动手造轮子:实现简单的 EventQueue
动手造轮子:实现简单的 EventQueue Intro 最近项目里有遇到一些并发的问题,想实现一个队列来将并发的请求一个一个串行处理,可以理解为使用消息队列处理并发问题,之前实现过一个简单的 Eve ...
- 动手造轮子:写一个日志框架
动手造轮子:写一个日志框架 Intro 日志框架有很多,比如 log4net / nlog / serilog / microsoft.extensions.logging 等,如何在切换日志框架的时 ...
- 动手造轮子:实现一个简单的依赖注入(二) --- 服务注册优化
动手造轮子:实现一个简单的依赖注入(二) --- 服务注册优化 Intro 之前实现的那版依赖注入框架基本可用,但是感觉还是不够灵活,而且注册服务和解析服务在同一个地方感觉有点别扭,有点职责分离不够. ...
- 动手造轮子:实现一个简单的依赖注入(一)
动手造轮子:实现一个简单的依赖注入(一) Intro 在上一篇文章中主要介绍了一下要做的依赖注入的整体设计和大概编程体验,这篇文章要开始写代码了,开始实现自己的依赖注入框架. 类图 首先来温习一下上次 ...
- 动手造轮子:实现一个简单的依赖注入(零)
动手造轮子:实现一个简单的依赖注入(零) Intro 依赖注入为我们写程序带来了诸多好处,在微软的 .net core 出来的同时也发布了微软开发的依赖注入框架 Microsoft.Extension ...
- python安装轮子_自己动手造“轮子”---python常用的几个方法
前言:由于工作内容的原因,经常需要些python脚本,久而久之,发现有一些方法经常用到,于是就自己动手编辑了一些常用的.大众的.通用的方法.小弟不才,但也希望能为开源做做贡献. 最后再附上代码哈: 一 ...
- 造轮子之图片轮播组件(swiper)
图片轮播是种很常见的场景和功能,一般移动网站首页的轮播 banner,商品详情页的商品图片等位置都会用到此功能 像这种常用的场景功能肯定是有人早就写好插件了的,所以遇到这种场景,一般都遵循以下三步: ...
最新文章
- 高级 Linux 命令精通指南
- /etc/rc.d 与 /etc/profile或者./.bash_profile的区别
- 一个算法对于某个输入的循环次数是可以事先估计出来的_结合各路博客的EM算法理解...
- oracle分歧处理,ORACLE 过程中容错处理
- 使用wsimport将wsdl生成java
- 数据结构与算法系列——排序(3)_折半插入排序
- 计算机可爱的企鹅教案,《可爱的企鹅》教学设计
- 什么是抓包?为什么要抓包?
- 你还记得当年高考时的样子吗?
- iOS 获取 WKWebView 里面图片
- python报错warning: install_lib: 'build\lib' does not exist -- no Python modules to install
- matlab使用plot画图点太密集
- pandas.pivot_table详解
- html网页打不开二级网页,遇到二级网页打不开怎么办 讲解二级页面打不开的处理方法...
- 转:德鲁克:管理的本质是激发善意和潜能
- 7-1 厘米换算英尺英寸 (JAVA)
- Unity3D引擎之Shader Forge应用
- 人工智能初学者指南:计算机视觉和图像识别
- 一起学习微信小程序api
- 华为mate10用的android版本,Android P明日适配华为Mate10
热门文章
- android mvvm 官方例子,详解Android的MVVM框架 - 数据绑定
- 协鑫集成“熄灯工厂”装上ET工业大脑,验证订单命中率可提高3.99%
- Linux学习笔记之一————什么是Linux及其应用领域
- java.lang.NoSuchFieldError: EMPTY_ORDERED_ITERATOR起因及解决办法
- 梦回编程- 由LD_LIBRARY_PATH引发JNI的理解
- 玩一玩微信公众号开发(一) 接入系统
- openresty 前端开发进阶一之http后端
- Android文件Apk下载变ZIP压缩包解决方案
- [禅悟人生]尊严非席, 不可卷起
- 好久没到这个地方来.