opt:提取配置重试冷却窗口
This commit is contained in:
parent
33c36bed10
commit
5712f567df
@ -17,37 +17,40 @@ public class CapFilter : SubscribeFilter
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="context"></param>
|
/// <param name="context"></param>
|
||||||
/// <returns></returns>
|
/// <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"]
|
if (!context.DeliverMessage.Headers["cap-msg-name"]
|
||||||
.IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe))
|
.IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe))
|
||||||
return Task.CompletedTask;
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
var argumentJsonStr = context.Arguments.ToJson();
|
var argumentJsonStr = context.Arguments.ToJson();
|
||||||
|
|
||||||
if (!string.IsNullOrEmpty(argumentJsonStr))
|
if (!string.IsNullOrEmpty(argumentJsonStr))
|
||||||
{
|
{
|
||||||
var hash = GetHash(argumentJsonStr);
|
var hash = context.DeliverMessage.Headers["sjzy-arguments-hash"];
|
||||||
context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash;
|
if (hash.IsNullOrEmpty()) // 计算参数hash
|
||||||
if (!string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}")))
|
{
|
||||||
throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数:{argumentJsonStr}");
|
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);
|
context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
|
||||||
if (!hash.IsNullOrEmpty())
|
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>
|
/// <summary>
|
||||||
@ -65,7 +68,11 @@ public class CapFilter : SubscribeFilter
|
|||||||
LogHelper.Warn(ex.ToString());
|
LogHelper.Warn(ex.ToString());
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
|
{
|
||||||
|
// 发飞书,异常继续上抛,进入重试机制
|
||||||
|
// tips:重试有冷却期(默认4分钟)
|
||||||
LogHelper.Error($"cap-msg-id:{messageId}\n{context.Exception}");
|
LogHelper.Error($"cap-msg-id:{messageId}\n{context.Exception}");
|
||||||
|
}
|
||||||
|
|
||||||
return Task.CompletedTask;
|
return Task.CompletedTask;
|
||||||
}
|
}
|
||||||
|
|||||||
@ -97,6 +97,8 @@ namespace SJZY.Track.Application
|
|||||||
options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval");
|
options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval");
|
||||||
options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount");
|
options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount");
|
||||||
options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter");
|
options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter");
|
||||||
|
options.FallbackWindowLookbackSeconds =
|
||||||
|
configuration.GetValue<int>("CAP:FallbackWindowLookbackSeconds");
|
||||||
options.EnableSubscriberParallelExecute = true;
|
options.EnableSubscriberParallelExecute = true;
|
||||||
|
|
||||||
options.UseDashboard(dashboardOptions =>
|
options.UseDashboard(dashboardOptions =>
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user