诺禾-EventBus/EventQueue 再思考

Posted 2021-02-18

最近把 Event 相关的逻辑做了一个重构，修正 EventStore，引入了 IEventHandlerFactory，重新设计了 Event 相关的组件

重构后的 Event
Event: 事情的笼统定义
EventHandler：事情处置器笼统定义
EventHandlerFactory：事情处置器工厂，用来依据事情类型获取事情处置器（新增）
EventPublisher：事情发布器，用于事情发布
EventSubscriber：事情订阅器，用于管理事情的订阅
EventSubscriptionManager：事情订阅管理器，在 EventSubscriber 的根底上增加了一个依据事情类型获取事情订阅器类型的办法
EventBus：事情总线，由 EventPubliser 和 EventSubscriber 组合而成，用来比拟便当的做事情发布和订阅
EventQueue：事情队列，希望某些音讯次第处置的时分能够思索用 EventQueue 的形式
EventStore：事情存储，事情的耐久化存储（在之前的版本里，EventStore 实践作用是一个 EventSubscriptionManager，在最近的版本更新中已修正)
以上 EventSubscriber 和 EventSubscriptionManager 普通不直接用，普通用 EventBus 来处置即可

EventHandlerFactory
这次引入了 EventHandlerFactory 用来笼统获取 EventHandler 的逻辑，原来的设计里是在处置 Event 的时分获取 EventHandler 的类型，然后从依赖注入框架中获取或创立新的 event handler 实例之后再调用 EventHandler 的 Handle 办法处置事情，有一些冗余

运用 EventHandlerFactory 之后就能够直接获取一个 EventHandler 实例汇合，详细是实例化还是从依赖注入中获取就由 EventHandlerFactory 来决议了，这样就能够对依赖注入很友好，关于基于内存的简单 EventBus 来说，在效劳注册之后就不需求再调用 Subscribe 去显式订阅了，由于再注册效劳的时分就曾经隐式完成了订阅的逻辑，这样实践就不需求 EventSubscriptionManager 来管理订阅了，订阅信息都在依赖注入框架内部，比方说 CounterEvent，要获取它的订阅信息，我只需求从依赖注入框架中获取 IEventHandler 的实例即可，实践就替代了原先 “EventStoreInMemory”，如今的 EventSubscriptionManagerInMemory

基于依赖注入的 EventHandlerFactory 定义：

public sealed class DependencyInjectionEventHandlerFactory : IEventHandlerFactory
{
private readonly IServiceProvider _serviceProvider;
public DependencyInjectionEventHandlerFactory(IServiceProvider serviceProvider = null)
{
_serviceProvider = serviceProvider ?? DependencyResolver.Current;
}
public ICollection GetHandlers(Type eventType)
{
var eventHandlerType = typeof(IEventHandler<>).MakeGenericType(eventType);
return _serviceProvider.GetServices(eventHandlerType).Cast().ToArray();
}
}
假如不运用依赖注入，也能够依据 IEventSubscriptionManager 订阅信息来完成：

public sealed class DefaultEventHandlerFactory : IEventHandlerFactory
{
private readonly IEventSubscriptionManager _subscriptionManager;
private readonly ConcurrentDictionary<Type, ICollection> _eventHandlers = new ConcurrentDictionary<Type, ICollection>();
private readonly IServiceProvider _serviceProvider;
public DefaultEventHandlerFactory(IEventSubscriptionManager subscriptionManager, IServiceProvider serviceProvider = null)
{
_subscriptionManager = subscriptionManager;
_serviceProvider = serviceProvider ?? DependencyResolver.Current;
}
public ICollection GetHandlers(Type eventType)
{
var eventHandlers = _eventHandlers.GetOrAdd(eventType, type =>
{
var handlerTypes = _subscriptionManager.GetEventHandlerTypes(type);
var handlers = handlerTypes
.Select(t => (IEventHandler)_serviceProvider.GetServiceOrCreateInstance(t))
.ToArray();
return handlers;
});
return eventHandlers;
}
}
EventQueue Demo
来看一下 EventQueue 的示例，示例基于 asp.net core 的，定义了一个 HostedService 来完成一个 EventConsumer 来消费 EventQueue 中的事情信息

EventConsumer 定义如下：

public class EventConsumer : BackgroundService
{
private readonly IEventQueue _eventQueue;
private readonly IEventHandlerFactory _eventHandlerFactory;
public EventConsumer(IEventQueue eventQueue, IEventHandlerFactory eventHandlerFactory)
{
_eventQueue = eventQueue;
_eventHandlerFactory = eventHandlerFactory;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
var queues = await _eventQueue.GetQueuesAsync();
if (queues.Count > 0)
{
await queues.Select(async q =>
{
var @event = await _eventQueue.DequeueAsync(q);
if (null != @event)
{
var handlers = _eventHandlerFactory.GetHandlers(@event.GetType());
if (handlers.Count > 0)
{
await handlers
.Select(h => h.Handle(@event))
.WhenAll()
;
}
}
})
.WhenAll()
;
}
await Task.Delay(1000, stoppingToken);
}
}
}
定义 PageViewEvent 和 PageViewEventHandler，用来记载和处置恳求访问记载

public class PageViewEvent : EventBase
{
}
public class PageViewEventHandler : EventHandlerBase<PageViewEvent>
{
public static int Count;
public override Task Handle(PageViewEvent @event)
{
Interlocked.Increment(ref Count);
return Task.CompletedTask;
}
}
事情很简单，事情处置也只是增加了 PageViewEventHandler 内定义的 Count。

效劳注册：

// 注册事情中心组件
// 会注册 EventBus、EventHandlerFactory、EventQueue 等
services.AddEvents()
// 注册 EventHanlder
.AddEventHandler<PageViewEvent, PageViewEventHandler>()
;
// 注册 EventQueuePubliser，默许注册的 IEventPublisher 是 EventBus
services.AddSingleton<IEventPublisher, EventQueuePublisher>();
// 注册 EventConsumer
services.AddHostedService();
事情发布，定义了一个中间件来发布 PageViewEvent，定义如下：

// pageView middleware
app.Use((context, next) =>
{
var eventPublisher = context.RequestServices.GetRequiredService();
eventPublisher.Publish(new PageViewEvent());
return next();
});
然后定义一个接口来获取上面定义的 PageViewEventHandler 中的 Count

[Route("api/[controller]")]
public class EventsController : ControllerBase
{
[HttpGet("pageViewCount")]
public IActionResult Count()
{
return Ok(new { Count = PageViewEventHandler.Count });
}
}
运转起来之后，访问几次接口，看上面的接口返回 Count 能否会增加，正常的话每访问一次接口就会增加 1，并发访问问题也不大，由于每个事情都是次第处置的，即便并发访问也没有关系，事情发布之后，在队列里都是次第处置的，这也就是引入事情队列的目的（仿佛上面的原子递增没什么用了...) 假如没看到了增加，稍等一会儿再访问试试，事情处置会迟到，但总会处置，毕竟是异步处置的，有些延迟很正常，而且上面我们还有一个 1s 的延迟