cap消费幂等

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/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)