动手造轮子：基于 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 的实现即可。
-
注册服务
services.AddSingleton<IEventBus, RedisEventBus>(); services.AddSingleton<IEventStore, EventStoreInRedis>(); -
注册
EventHandlerservices.AddSingleton<NoticeViewEventHandler>(); -
订阅事件
eventBus.Subscribe<NoticeViewEvent, NoticeViewEventHandler>(); -
发布事件
[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 });
});
}
}
}
网友评论