diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs new file mode 100644 index 00000000..fbee07b0 --- /dev/null +++ b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs @@ -0,0 +1,109 @@ +using System.Globalization; +using DotNetCore.CAP.Filter; +using SJZY.Expand.ABP.Core; +using SJZY.Expand.ABP.Core.Logger; +using SJZY.Expand.ABP.Core.PublicMethords; + +namespace SJZY.Track.Application.EventHandlers; + +public class CAPFilter : SubscribeFilter +{ + const string _ERROR_CODE = "Error599"; + + /// + /// 执行前 + /// + /// + /// + public override Task OnSubscribeExecutingAsync(ExecutingContext context) + { + var argumentJsonStr = context.Arguments.ToJson(); + + context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff"); + context.DeliverMessage.Headers["sjzy-arguments"] = argumentJsonStr; + + if (!string.IsNullOrEmpty(argumentJsonStr)) + { + var hash = argumentJsonStr.GetHash(); + using (RedisHelper.Client.Lock("Filter:" + hash, 10)) + { + var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExcuteFilter:{hash}")); + if (isSkip) + throw new BusinessException(_ERROR_CODE, $"检测到重复消费,消费参数:{argumentJsonStr}"); + + RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10, + TimeSpan.FromDays(1).Seconds); // 1天内不允许一模一样的参数重复消费 + } + } + + return Task.CompletedTask; + } + + /// + /// 执行后 + /// + /// + /// + public override Task OnSubscribeExecutedAsync(ExecutedContext context) + { + context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); + context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); + context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); + context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); + + BuildLog(receiveTime, receiveName, arguments, false, messageId); + + return Task.CompletedTask; + } + + + /// + /// 消费异常事件 + /// + /// + /// + public override Task OnSubscribeExceptionAsync(ExceptionContext context) + { + context.ExceptionHandled = true; + + if (context.Exception is BusinessException ex && ex.Code == _ERROR_CODE) + LogHelper.Warn(ex.ToString()); + else + LogHelper.Error(context.Exception.ToString()); + + + context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); + context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); + context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); + context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); + + BuildLog(receiveTime, receiveName, arguments, true, messageId); + + return Task.CompletedTask; + } + + + private void BuildLog(string receiveTime, string receiveName, string arguments, bool isFromExecptionFliter, + string messageId) + { + var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture); + var useTime = DateTime.Now - parsedDate; + + var log = $@" +|-消息编号:{messageId} +|-消费者名:{receiveName} +|-跳过执行:{(isFromExecptionFliter ? "是" : "否")} +|-执行时间:{receiveTime} +|-消费耗时:{useTime.Milliseconds}毫秒 +|-请求参数:{arguments}"; + + if (useTime.Minutes > 20) + { + LogHelper.Error(log); + } + else + { + //LogHelper.Info(log); + } + } +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index b2b2cd06..88062bb8 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -6,6 +6,7 @@ using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract; +using SJZY.Track.Application.EventHandlers; using SJZY.Track.Application.Localization; using SJZY.UserCenter.Contracts; using SJZY.UserCenter.SharedModel; @@ -90,7 +91,7 @@ namespace SJZY.Track.Application { dashboardOptions.PathBase = System.Diagnostics.Debugger.IsAttached ? "" - : configuration["Swagger:JsonUrlPrefix"]; + : configuration["Swagger:JsonUrlPrefix"]!; }); options.FailedThresholdCallback = faileCalback => { @@ -99,15 +100,15 @@ namespace SJZY.Track.Application LogHelper.Error($@" |-CAP失败回调事件: |-处理类型:{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数,需要开发人员介入" : "生产消息推送失败")} - |-头部信息:{faileCalback?.Message?.Headers?.ToJson()} - |-消息内容:{((JsonElement)faileCalback?.Message?.Value).GetRawText()}"); + |-头部信息:{faileCalback.Message.Headers.ToJson()} + |-消息内容:{((JsonElement?)faileCalback.Message.Value)?.GetRawText()}"); } catch (Exception ex) { LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex); } }; - }); + }).AddSubscribeFilter(); } public override void PreConfigureServices(ServiceConfigurationContext context)