動手造輪子：實作簡單的 EventQueue
Intro
最近專案里有遇到一些并發的問題，想實作一個佇列來將并發的請求一個一個串行處理，可以理解為使用訊息佇列處理并發問題，之前實作過一個簡單的 EventBus，于是想在 EventBus 的基礎上改造一下，加一個佇列，改造成類似訊息佇列的處理模式，訊息的處理（Consumer）直接使用 .netcore 里的 IHostedService 來實作了一個簡單的后臺任務處理，
初步設計
- Event 抽象的事件
- EventHandler 處理 Event 的方法
- EventStore 保存訂閱 Event 的 EventHandler
- EventQueue 保存 Event 的佇列
- EventPublisher 發布 Event
- EventConsumer 處理 Event 佇列里的 Event
- EventSubscriptionManager 管理訂閱 Event 的 EventHandler
實作代碼
EventBase 定義了基本事件資訊，事件發生時間以及事件的id:
public abstract class EventBase
{
[JsonProperty]
public DateTimeOffset EventAt { get; private set; }
[JsonProperty]
public string EventId { get; private set; }
protected EventBase()
{
this.EventId = GuidIdGenerator.Instance.NewId();
this.EventAt = DateTimeOffset.UtcNow;
}
[JsonConstructor]
public EventBase(string eventId, DateTimeOffset eventAt)
{
this.EventId = eventId;
this.EventAt = eventAt;
}
}
EventHandler 定義：
public interface IEventHandler
{
Task Handle(IEventBase @event);
}
public interface IEventHandler<in TEvent> : IEventHandler where TEvent : IEventBase
{
Task Handle(TEvent @event);
}
public class EventHandlerBase<TEvent> : IEventHandler<TEvent> where TEvent : EventBase
{
public virtual Task Handle(TEvent @event)
{
return Task.CompletedTask;
}
public Task Handle(IEventBase @event)
{
return Handle(@event as TEvent);
}
}
EventStore：
public class EventStore
{
private readonly Dictionary<Type, Type> _eventHandlers = new Dictionary<Type, Type>();
public void Add<TEvent, TEventHandler>() where TEventHandler : IEventHandler<TEvent> where TEvent : EventBase
{
_eventHandlers.Add(typeof(TEvent), typeof(TEventHandler));
}
public object GetEventHandler(Type eventType, IServiceProvider serviceProvider)
{
if (eventType == null || !_eventHandlers.TryGetValue(eventType, out var handlerType) || handlerType == null)
{
return null;
}
return serviceProvider.GetService(handlerType);
}
public object GetEventHandler(EventBase eventBase, IServiceProvider serviceProvider) =>
GetEventHandler(eventBase.GetType(), serviceProvider);
public object GetEventHandler<TEvent>(IServiceProvider serviceProvider) where TEvent : EventBase =>
GetEventHandler(typeof(TEvent), serviceProvider);
}
EventQueue 定義：
public class EventQueue
{
private readonly ConcurrentDictionary<string, ConcurrentQueue<EventBase>> _eventQueues =
new ConcurrentDictionary<string, ConcurrentQueue<EventBase>>();
public ICollection<string> Queues => _eventQueues.Keys;
public void Enqueue<TEvent>(string queueName, TEvent @event) where TEvent : EventBase
{
var queue = _eventQueues.GetOrAdd(queueName, q => new ConcurrentQueue<EventBase>());
queue.Enqueue(@event);
}
public bool TryDequeue(string queueName, out EventBase @event)
{
var queue = _eventQueues.GetOrAdd(queueName, q => new ConcurrentQueue<EventBase>());
return queue.TryDequeue(out @event);
}
public bool TryRemoveQueue(string queueName)
{
return _eventQueues.TryRemove(queueName, out _);
}
public bool ContainsQueue(string queueName) => _eventQueues.ContainsKey(queueName);
public ConcurrentQueue<EventBase> this[string queueName] => _eventQueues[queueName];
}
EventPublisher:
public interface IEventPublisher
{
Task Publish<TEvent>(string queueName, TEvent @event)
where TEvent : EventBase;
}
public class EventPublisher : IEventPublisher
{
private readonly EventQueue _eventQueue;
public EventPublisher(EventQueue eventQueue)
{
_eventQueue = eventQueue;
}
public Task Publish<TEvent>(string queueName, TEvent @event)
where TEvent : EventBase
{
_eventQueue.Enqueue(queueName, @event);
return Task.CompletedTask;
}
}
EventSubscriptionManager:
public interface IEventSubscriptionManager
{
void Subscribe<TEvent, TEventHandler>()
where TEvent : EventBase
where TEventHandler : IEventHandler<TEvent>;
}
public class EventSubscriptionManager : IEventSubscriptionManager
{
private readonly EventStore _eventStore;
public EventSubscriptionManager(EventStore eventStore)
{
_eventStore = eventStore;
}
public void Subscribe<TEvent, TEventHandler>()
where TEvent : EventBase
where TEventHandler : IEventHandler<TEvent>
{
_eventStore.Add<TEvent, TEventHandler>();
}
}
EventConsumer:
public class EventConsumer : BackgroundService
{
private readonly EventQueue _eventQueue;
private readonly EventStore _eventStore;
private readonly int maxSemaphoreCount = 256;
private readonly IServiceProvider _serviceProvider;
private readonly ILogger _logger;
public EventConsumer(EventQueue eventQueue, EventStore eventStore, IConfiguration configuration, ILogger<EventConsumer> logger, IServiceProvider serviceProvider)
{
_eventQueue = eventQueue;
_eventStore = eventStore;
_logger = logger;
_serviceProvider = serviceProvider;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
using (var semaphore = new SemaphoreSlim(Environment.ProcessorCount, maxSemaphoreCount))
{
while (!stoppingToken.IsCancellationRequested)
{
var queues = _eventQueue.Queues;
if (queues.Count > 0)
{
await Task.WhenAll(
queues
.Select(async queueName =>
{
if (!_eventQueue.ContainsQueue(queueName))
{
return;
}
try
{
await semaphore.WaitAsync(stoppingToken);
//
if (_eventQueue.TryDequeue(queueName, out var @event))
{
var eventHandler = _eventStore.GetEventHandler(@event, _serviceProvider);
if (eventHandler is IEventHandler handler)
{
_logger.LogInformation(
"handler {handlerType} begin to handle event {eventType}, eventId: {eventId}, eventInfo: {eventInfo}",
eventHandler.GetType().FullName, @event.GetType().FullName,
@event.EventId, JsonConvert.SerializeObject(@event));
try
{
await handler.Handle(@event);
}
catch (Exception e)
{
_logger.LogError(e, "event {eventId} handled exception", @event.EventId);
}
finally
{
_logger.LogInformation("event {eventId} handled", @event.EventId);
}
}
else
{
_logger.LogWarning(
"no event handler registered for event {eventType}, eventId: {eventId}, eventInfo: {eventInfo}",
@event.GetType().FullName, @event.EventId,
JsonConvert.SerializeObject(@event));
}
}
}
catch (Exception ex)
{
_logger.LogError(ex, "error running EventConsumer");
}
finally
{
semaphore.Release();
}
})
);
}
await Task.Delay(50, stoppingToken);
}
}
}
}
為了方便使用定義了一個 Event 擴展方法：
public static IServiceCollection AddEvent(this IServiceCollection services)
{
services.TryAddSingleton<EventStore>();
services.TryAddSingleton<EventQueue>();
services.TryAddSingleton<IEventPublisher, EventPublisher>();
services.TryAddSingleton<IEventSubscriptionManager, EventSubscriptionManager>();
services.AddSingleton<IHostedService, EventConsumer>();
return services;
}
使用示例
定義 PageViewEvent 記錄請求資訊：
public class PageViewEvent : EventBase
{
public string Path { get; set; }
}
這里作為示例只記錄了請求的Path資訊，實際使用可以增加更多需要記錄的資訊
定義 PageViewEventHandler，處理 PageViewEvent
public class PageViewEventHandler : EventHandlerBase<PageViewEvent>
{
private readonly ILogger _logger;
public PageViewEventHandler(ILogger<PageViewEventHandler> logger)
{
_logger = logger;
}
public override Task Handle(PageViewEvent @event)
{
_logger.LogInformation($"handle pageViewEvent: {JsonConvert.SerializeObject(@event)}");
return Task.CompletedTask;
}
}
這個 handler 里什么都沒做只是輸出一個日志
這個示例專案定義了一個記錄請求路徑的事件以及一個發布請求記錄事件的中間件
// 發布 Event 的中間件
app.Use(async (context, next) =>
{
var eventPublisher = context.RequestServices.GetRequiredService<IEventPublisher>();
await eventPublisher.Publish("pageView", new PageViewEvent() { Path = context.Request.Path.Value });
await next();
});
Startup 配置：
public void ConfigureServices(IServiceCollection services)
{
// ...
services.AddEvent();
services.AddSingleton<PageViewEventHandler>();// 注冊 Handler
}
// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
public void Configure(IApplicationBuilder app, IHostingEnvironment env, IEventSubscriptionManager eventSubscriptionManager)
{
eventSubscriptionManager.Subscribe<PageViewEvent, PageViewEventHandler>();
app.Use(async (context, next) =>
{
var eventPublisher = context.RequestServices.GetRequiredService<IEventPublisher>();
await eventPublisher.Publish("pageView", new PageViewEvent() { Path = context.Request.Path.Value });
await next();
});
// ...
}
使用效果:
More
注：只是一個初步設計，基本可以實作功能，還是有些不足，實際應用的話還有一些要考慮的事情
Consumer訊息邏輯，現在的實作有些問題，我們的應用場景目前比較簡單還可以滿足，如果事件比較多就會而且每個事件可能處理需要的時間長短不一樣，會導致在一個批次中執行的Event中已經完成的事件要等待其他還沒完成的事件完成之后才能繼續取下一個事件，理想的消費模式應該是各個佇列相互獨立，在同一個佇列中保持順序消費即可- 上面示例的
EventStore的實作只是簡單的實作了一個事件一個Handler的處理情況，實際業務場景中很可能會有一個事件需要多個Handler的情況 - 這個實作是基于記憶體的，如果要在分布式場景下使用就不適用了，需要自己實作一下基于redis或者資料庫的以滿足分布式的需求
- and more...
上面所有的代碼可以在 Github 上獲取，示例專案 Github 地址：https://github.com/WeihanLi/AspNetCorePlayground/tree/master/TestWebApplication
Reference
- https://github.com/WeihanLi/AspNetCorePlayground/tree/master/TestWebApplication/Event
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/89731.html
標籤：.NET Core