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 /// /// /// - 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; } /// @@ -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("CAP:FailedRetryInterval"); options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); + options.FallbackWindowLookbackSeconds = + configuration.GetValue("CAP:FallbackWindowLookbackSeconds"); options.EnableSubscriberParallelExecute = true; options.UseDashboard(dashboardOptions =>