Spring @StreamListener：使用指数退避进行无限重试答案

【问题标题】：Spring @StreamListener: Infinite retries with exponential backoffSpring @StreamListener：使用指数退避进行无限重试
【发布时间】：2021-07-04 12:07:40
【问题描述】：

我正在尝试将我的消费者配置为使用指数退避，其中消息将被处理固定次数的尝试，并在其中应用退避期。但我没有得到预期的行为。

这是我的 Java 代码：

@EnableBinding({
        MessagingConfiguration.EventTopic.class
})
public class MessagingConfiguration {

    public interface EventTopic {
        String INPUT = "events-channel";

        @Input(INPUT)
        @Nonnull
        SubscribableChannel input();
    }
}
  
@StreamListener(MessagingConfiguration.EventTopic.INPUT))
void handle(@Nonnull Message<Event> event) {
    throw new RuntimeException("FAILING!");
}

如果我尝试下一个配置：

spring.cloud.stream:
  bindings:
    events-channel:
      content-type: application/json
      destination: event-develop
      group: group-event-service
      consumer:
        max-attempts: 2

在所有重试 (20*) 后，我收到此消息：

Backoff FixedBackOff{interval=0, currentAttempts=10, maxAttempts=9} exhausted for ConsumerRecord(...

2 (consumer.max-attempts) * 10 (FixedBackOff.currentAttempts) = 20* 次重试

所有这些重试都会延迟 1 秒（默认退避期）

如果我将配置更改为：

spring.cloud.stream:
  bindings:
    events-channel:
      content-type: application/json
      destination: event-develop
      group: group-event-service
      consumer:
        max-attempts: 8
        #Times in milliseconds
        back-off-initial-interval: 1000
        back-off-max-interval: 60000
        back-off-multiplier: 2

在 8 次重试期间很好地应用了退避期 (max-attempts)，但是当 8 次重试完成时，新的重试周期开始无限期地阻止主题。

在下一个版本中，也许我会实现更复杂的错误处理系统，但现在我只需要在重试后丢弃消息并获取下一个。

我做错了什么？

我在这里阅读了很多问题/答案，官方文档和互联网上的一些教程，但我没有找到避免重试无限循环的解决方案。

P.S.：我正在使用 spring-cloud-stream (3.1.1) 和 spring-kafka (2.6.6)

【问题讨论】：

标签： spring-kafka spring-cloud-stream spring-retry

【解决方案1】：

这是因为现在默认情况下，侦听器容器配置为 SeekToCurrentErrorHandler，尝试 10 次。

这意味着您正在复合重试。

您可以使用ListenerContainerCustomizer @Bean 注入适当配置的SeekToCurrentErrorHandler。

建议不要在两个地方都配置重试；要么删除绑定配置并将其替换为适当配置的错误处理程序，要么将错误处理程序更改为不重试。

【讨论】：

谢谢你，Gary，你让我走对了。