動(dòng)手造輪子：基于 Redis 實(shí)現(xiàn) EventBus

Intro

上次我們?cè)炝艘粋€(gè)簡(jiǎn)單的基于內(nèi)存的 EventBus，但是如果要跨系統(tǒng)的話就不合適了，所以有了這篇基于 Redis 的 EventBus 探索。本文的實(shí)現(xiàn)是基于 StackExchange.Redis 來(lái)實(shí)現(xiàn)。

RedisEventStore?實(shí)現(xiàn)

既然要實(shí)現(xiàn)跨系統(tǒng)的 EventBus 再使用基于內(nèi)存的 EventStore 自然不行，因此這里基于 Redis 設(shè)計(jì)了一個(gè) EventStoreInRedis ,基于 redis 的 Hash 來(lái)實(shí)現(xiàn)，以 Event 的 EventKey 作為 fieldName，以 Event 對(duì)應(yīng)的 EventHandler 作為 Value。

EventStoreInRedis 實(shí)現(xiàn)：

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 的擴(kuò)展的實(shí)現(xiàn) https://github.com/WeihanLi/WeihanLi.Redis

RedisEventBus?實(shí)現(xiàn)

RedisEventBus 是基于 Redis 的 PUB/SUB 實(shí)現(xiàn)的，實(shí)現(xiàn)的感覺(jué)還有一些小問(wèn)題，我想確保每個(gè)客戶端注冊(cè)的時(shí)候每個(gè) EventHandler 即使多次注冊(cè)也只注冊(cè)一次，但是還沒(méi)找到一個(gè)好的實(shí)現(xiàn)，如果你有什么想法歡迎指出，和我一起交流。具體的實(shí)現(xiàn)細(xì)節(jié)如下：

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; } }

使用示例：

使用起來(lái)大體上和上一篇使用一致，只是在初始化注入服務(wù)的時(shí)候，我們需要把 IEventBus 和 IEventStore 替換為對(duì)應(yīng) Redis 的實(shí)現(xiàn)即可。

1. 注冊(cè)服務(wù)

services.AddSingleton<IEventBus, RedisEventBus>(); services.AddSingleton<IEventStore, EventStoreInRedis>();

2. 注冊(cè)?EventHandler

services.AddSingleton<NoticeViewEventHandler>();

3. 訂閱事件

eventBus.Subscribe<NoticeViewEvent, NoticeViewEventHandler>();

4. 發(fā)布事件

[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

如果要實(shí)現(xiàn)基于消息隊(duì)列的事件處理，需要注意，消息可能會(huì)重復(fù)，可能會(huì)需要在事件處理中注意一下業(yè)務(wù)的冪等性或者對(duì)消息對(duì)一個(gè)去重處理。

我在使用 Redis 的事件處理中使用了一個(gè)基于 Redis 原子遞增的特性設(shè)計(jì)的一個(gè)防火墻，從而實(shí)現(xiàn)一段時(shí)間內(nèi)某一個(gè)消息id只會(huì)被處理一次，實(shí)現(xiàn)源碼：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

總結(jié)

以上是生活随笔為你收集整理的动手造轮子：基于 Redis 实现 EventBus的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。