From 5712f567dfbf9d5168f2e9b6a3828baefd0fe1d2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com>
Date: Mon, 27 May 2024 17:00:39 +0800
Subject: [PATCH] =?UTF-8?q?opt=EF=BC=9A=E6=8F=90=E5=8F=96=E9=85=8D?=
 =?UTF-8?q?=E7=BD=AE=E9=87=8D=E8=AF=95=E5=86=B7=E5=8D=B4=E7=AA=97=E5=8F=A3?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/SJZY.Track.Application/CAP/CAPFilter.cs   | 35 +++++++++++--------
 .../SjzyTrackApplicationModule.cs             |  2 ++
 2 files changed, 23 insertions(+), 14 deletions(-)

diff --git a/src/SJZY.Track.Application/CAP/CAPFilter.cs b/src/SJZY.Track.Application/CAP/CAPFilter.cs
index 110395fc..6522599b 100644
--- a/src/SJZY.Track.Application/CAP/CAPFilter.cs
+++ b/src/SJZY.Track.Application/CAP/CAPFilter.cs
@@ -17,37 +17,40 @@ public class CapFilter : SubscribeFilter
     /// </summary>
     /// <param name="context"></param>
     /// <returns></returns>
-    public override Task OnSubscribeExecutingAsync(ExecutingContext context)
+    public override async Task OnSubscribeExecutingAsync(ExecutingContext context)
     {
-        if (!context.DeliverMessage.Headers["sjzy-arguments-hash"].IsNullOrEmpty()) return Task.CompletedTask;
-
         // 轨迹新增、更新要做幂等
         if (!context.DeliverMessage.Headers["cap-msg-name"]
                 .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe))
-            return Task.CompletedTask;
+        {
+            return;
+        }
 
         var argumentJsonStr = context.Arguments.ToJson();
 
         if (!string.IsNullOrEmpty(argumentJsonStr))
         {
-            var hash = GetHash(argumentJsonStr);
-            context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash;
-            if (!string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}")))
-                throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}");
-        }
+            var hash = context.DeliverMessage.Headers["sjzy-arguments-hash"];
+            if (hash.IsNullOrEmpty()) // 计算参数hash
+            {
+                hash = GetHash(argumentJsonStr);
+                context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash;
+            }
 
-        return Task.CompletedTask;
+            if (await RedisHelper.Client.ExistsAsync($"CapExecuteFilter:{hash}"))
+            {
+                throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}");
+            }
+        }
     }
 
-    public override Task OnSubscribeExecutedAsync(ExecutedContext context)
+    public override async Task OnSubscribeExecutedAsync(ExecutedContext context)
     {
         context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
         if (!hash.IsNullOrEmpty())
         {
-            RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10, TimeSpan.FromMinutes(10).Seconds);
+            await RedisHelper.Client.SetAsync($"CapExecuteFilter:{hash}", 1, TimeSpan.FromMinutes(10).Seconds);
         }
-
-        return Task.CompletedTask;
     }
 
     /// <summary>
@@ -65,7 +68,11 @@ public class CapFilter : SubscribeFilter
             LogHelper.Warn(ex.ToString());
         }
         else
+        {
+            // 发飞书，异常继续上抛，进入重试机制
+            // tips：重试有冷却期（默认4分钟）
             LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}");
+        }
 
         return Task.CompletedTask;
     }
diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
index efa5aa41..10394daa 100644
--- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
+++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
@@ -97,6 +97,8 @@ namespace SJZY.Track.Application
                 options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval");
                 options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount");
                 options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter");
+                options.FallbackWindowLookbackSeconds =
+                    configuration.GetValue<int>("CAP:FallbackWindowLookbackSeconds");
                 options.EnableSubscriberParallelExecute = true;
 
                 options.UseDashboard(dashboardOptions =>