This commit is contained in:
@@ -29,6 +29,17 @@ var acquireConcurrencyScript = redis.NewScript(`
|
||||
return current
|
||||
`)
|
||||
|
||||
var releaseConcurrencyScript = redis.NewScript(`
|
||||
local current = tonumber(redis.call('GET', KEYS[1]) or '0') or 0
|
||||
if current <= 0 then
|
||||
if redis.call('EXISTS', KEYS[1]) == 1 then
|
||||
redis.call('SET', KEYS[1], 0, 'KEEPTTL')
|
||||
end
|
||||
return 0
|
||||
end
|
||||
return redis.call('DECR', KEYS[1])
|
||||
`)
|
||||
|
||||
const pollingFallbackOperationTimeout = 5 * time.Second
|
||||
|
||||
// pollingFallbackContext 创建轮询兜底操作使用的独立短超时上下文。
|
||||
@@ -101,7 +112,7 @@ func (b *PollingBase) releaseConcurrency(_ context.Context, taskType string) {
|
||||
defer cancel()
|
||||
|
||||
currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType)
|
||||
if err := b.redis.Decr(ctx, currentKey).Err(); err != nil {
|
||||
if err := releaseConcurrencyScript.Run(ctx, b.redis, []string{currentKey}).Err(); err != nil {
|
||||
b.logger.Warn("释放并发计数失败", zap.String("task_type", taskType), zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user