Merge branch 'S02-轨迹中台第二部分' into dev

src/SJZY.Track.Application.Contract/Events/TrackUpsertEvent.cs

@@ -38,4 +38,14 @@ public class TrackUpsertEvent : UpsertTrackInput
     /// 已经处理过的订单
     /// </summary>
     public List<int> Visited { get; set; } = new();
+
+    /// <summary>
+    /// 原轨迹内容
+    /// </summary>
+    public string? OriginTrackContent { get; set; }
+
+    /// <summary>
+    /// 原轨迹报文
+    /// </summary>
+    public string? OriginTrackData { get; set; }
 }

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";
+
+    /// <summary>
+    /// 执行前
+    /// </summary>
+    /// <param name="context"></param>
+    /// <returns></returns>
+    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;
+    }
+
+    /// <summary>
+    /// 执行后
+    /// </summary>
+    /// <param name="context"></param>
+    /// <returns></returns>
+    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;
+    }
+
+
+    /// <summary>
+    /// 消费异常事件
+    /// </summary>
+    /// <param name="context"></param>
+    /// <returns></returns>
+    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);
+        }
+    }
+}

src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs

@@ -95,6 +95,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
         TrackUpsertEvent @event)
     {
         var trackUpsertEvent = @event.DeepCopy();
+        trackUpsertEvent.OriginTrackData = JsonConvert.SerializeObject(trackToken);
         var paramConfig = paramWithContentConfig.ParamConfig;
         var contentConfigs = paramWithContentConfig.ContentConfigs;
         TrackPullContentConfigDto? matchedContentConfig = null;
@@ -123,6 +124,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
             var trackContent = trackToken[paramConfig.TrackContentName]?.Value<string>();
             if (!trackContent.IsNullOrWhiteSpace())
             {
+                trackUpsertEvent.OriginTrackContent = trackContent;
                 Debug.Assert(trackContent != null, nameof(trackContent) + " != null");
                 matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackContentValue == trackContent);
 

src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs

@@ -132,6 +132,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
                 TrackTime = @event.TrackTime!.Value,
                 TrackTimeZone = @event.TrackTimeZone!.Value,
                 IsCanDelete = trackContent.IsCanDelete,
+                OriginTrackContent = @event.OriginTrackContent,
+                OriginTrackData = @event.OriginTrackData,
                 CreateUser = CurrentUserRequest.CurrentUser.Name,
                 CreateTime = DateTime.Now
             };

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()}
+    |-头部信息：{faileCalback.Message.Headers.ToJson()}
-   	|-消息内容：{((JsonElement)faileCalback?.Message?.Value).GetRawText()}");
+   	|-消息内容：{((JsonElement?)faileCalback.Message.Value)?.GetRawText()}");
                     }
                     catch (Exception ex)
                     {
                         LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex);
                     }
                 };
-            });
+            }).AddSubscribeFilter<CAPFilter>();
         }
 
         public override void PreConfigureServices(ServiceConfigurationContext context)

src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/TrackDetailsIndex.cs

@@ -52,11 +52,6 @@ public class TrackDetailsIndex : IdBaseIndex
     /// </summary>
     public bool IsCompleted { get; set; }
 
-    /// <summary>
-    /// 原轨迹状态
-    /// </summary>
-    public string? OriginTrackStatus { get; set; }
-
     /// <summary>
     /// 原轨迹内容
     /// </summary>