结合 AOP 轻松处理事件发布处理日志
Intro
前段时间，实现了 EventBus 以及 EventQueue 基于 Event 的事件处理，但是没有做日志（EventLog）相关的部分，原本想增加两个接口， 处理事件发布日志和事件处理日志，最近用了 AOP 的思想处理了 EntityFramework 的数据变更自动审计，于是想着事件日志也用 AOP 的思想来实现，而且可能用 AOP 来处理可能会更好一些，最近自己造了一个 AOP 的轮子 —— FluentAspects，下面的示例就以它来演示了，你也可以换成自己喜欢的 AOP 组件，思想是类似的
事件发布日志
事件发布日志只需要拦截事件发布的方法调用即可，在发布事件时进行拦截，在拦截器中根据需要进行日志记录即可
事件发布者接口定义：
public interface IEventPublisher
{
/// <summary>
/// publish an event
/// </summary>
/// <typeparam name="TEvent">event type</typeparam>
/// <param name="event">event data</param>
/// <returns>whether the operation succeed</returns>
bool Publish<TEvent>(TEvent @event) where TEvent : class, IEventBase; /// <summary>
/// publish an event async
/// </summary>
/// <typeparam name="TEvent">event type</typeparam>
/// <param name="event">event data</param>
/// <returns>whether the operation succeed</returns>
Task<bool> PublishAsync<TEvent>(TEvent @event) where TEvent : class, IEventBase;
}
事件发布日志拦截器：
public class EventPublishLogInterceptor : AbstractInterceptor
{
public override async Task Invoke(IInvocation invocation, Func<Task> next)
{
Console.WriteLine("-------------------------------");
Console.WriteLine($"Event publish begin, eventData:{invocation.Arguments.ToJson()}");
var watch = Stopwatch.StartNew();
try
{
await next();
}
catch (Exception ex)
{
Console.WriteLine($"Event publish exception({ex})");
}
finally
{
watch.Stop();
Console.WriteLine($"Event publish complete, elasped:{watch.ElapsedMilliseconds} ms");
}
Console.WriteLine("-------------------------------");
}
}
事件处理日志
事件处理器接口定义：
public interface IEventHandler
{
Task Handle(object eventData);
}
事件处理日志拦截器定义：
public class EventHandleLogInterceptor : IInterceptor
{
public async Task Invoke(IInvocation invocation, Func<Task> next)
{
Console.WriteLine("-------------------------------");
Console.WriteLine($"Event handle begin, eventData:{invocation.Arguments.ToJson()}");
var watch = Stopwatch.StartNew();
try
{
await next();
}
catch (Exception ex)
{
Console.WriteLine($"Event handle exception({ex})");
}
finally
{
watch.Stop();
Console.WriteLine($"Event handle complete, elasped:{watch.ElapsedMilliseconds} ms");
}
Console.WriteLine("-------------------------------");
}
}
AOP 配置
Host.CreateDefaultBuilder(args)
.ConfigureWebHostDefaults(builder =>
{
builder.UseStartup<Startup>();
})
.UseFluentAspectServiceProviderFactory(options =>
{
// 拦截器配置 // 拦截 `IEventPublisher` 日志，注册事件发布日志拦截器
options
.InterceptType<IEventPublisher>()
.With<EventPublishLogInterceptor>(); // 拦截 `IEventHandler`，注册事件处理日志拦截器
options.InterceptType<IEventHandler>()
.With<EventHandleLogInterceptor>();
}, builder =>
{
// 默认使用默认实现来生成代理，现在提供了 Castle 和 AspectCore 的扩展，也可以自己扩展实现自定义代理生成方式
// 取消注释使用 Castle 来生成代理
//builder.UseCastleProxy();
}, t => t.Namespace?.StartsWith("WeihanLi") == false // 要忽略的类型断言
)
.Build()
.Run();
More
事件发布示例，定义了一个发布事件的中间件：
// pageView middleware
app.Use((context, next) =>
{
var eventPublisher = context.RequestServices
.GetRequiredService<IEventPublisher>();
eventPublisher.Publish(new PageViewEvent()
{
Path = context.Request.Path.Value,
}); return next();
});
事件处理示例是用一个消息队列的模式来处理的，示例和前面的事件的文章类似，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);
}
}
}
完整的示例代码可以从https://github.com/WeihanLi/WeihanLi.Common/blob/dev/samples/AspNetCoreSample 获取
