弹簧云卡夫卡流消费者重试机制

Question

下面这些属性的含义是什么?我如何使用它们？

spring.cloud.stream.bindings.atcommnity.consumer.maxAttempts=5
spring.cloud.stream.bindings.atcommnity.consumer.backOffInitialInterval=1000
spring.cloud.stream.bindings.atcommnity.consumer.backOffMaxInterval=2000000
spring.cloud.stream.bindings.atcommnity.consumer.backOffMultiplier=2.0
spring.cloud.stream.bindings.atcommnity.consumer.batch-mode=false

Answer 1

退避将从backOffInitialInterval开始，然后下一次尝试将乘以backOffMultiplier，但不会超过backOffMaxInterval。

currentInterval = Math.min(backOffInitialInterval * Math.pow(backOffMultiplier, retryNum), backOffMaxInterval)

就你而言，情况将是：

1000 -> 2000 -> 4000 -> 8000 -> 16000

Answer 2

以上亚历克斯的回答是对这个问题的完美解释。这里只是为了增加我的知识和分享一些额外的发现。我正在从代码中提供实际的实现。

private BackOff createBackOff(
        final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {

    int maxAttempts = extendedConsumerProperties.getMaxAttempts();
    if (maxAttempts < 2) {
        return new FixedBackOff(0L, 0L);
    }
    int initialInterval = extendedConsumerProperties.getBackOffInitialInterval();
    double multiplier = extendedConsumerProperties.getBackOffMultiplier();
    int maxInterval = extendedConsumerProperties.getBackOffMaxInterval();
    ExponentialBackOff backOff = new ExponentialBackOff(initialInterval, multiplier);
    backOff.setMaxInterval(maxInterval);
    long maxElapsed = extendedConsumerProperties.getBackOffInitialInterval();
    double accum = maxElapsed;
    for (int i = 1; i < maxAttempts - 1; i++) {
        accum = accum * multiplier;
        if (accum > maxInterval) {
            accum = maxInterval;
        }
        maxElapsed += accum;
    }
    backOff.setMaxElapsedTime(maxElapsed);
    return backOff;
}

类名KafkaMessageChannelBinder