当达到消费者 maxAttempts 时会发生什么？

Question

使用以下配置和场景，当达到 maxAttempts 时会发生什么情况？

Spring 具有 Kafka 绑定和以下属性的 Cloud Stream：

• spring.cloud.stream.bindings.input.consumer.maxAttempts=3
• spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset=真
• spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOnError=假
• spring.cloud.stream.kafka.bindings.input.consumer.enableDlq=假

场景如下：

• 消费者通过@StreamListener 注解接收消息负载
• 在从带注释的方法返回之前，消费者尝试将消息保存在数据库中
• 数据库已关闭，@StreamListener 注解方法抛出运行时异常

我看到的行为是消费者重试消息直到达到 maxAttempts 限制。然后在我重新启动服务之前什么也没有发生。重新启动后，消息将被重新使用。

如果数据库在达到 maxAttempts 后再次可用会怎样？重启服务是我唯一的选择吗？有没有办法将 maxAttempts 设置为无穷大？

我怀疑我没有完全理解这种行为

Answer 1

这确实是预期的行为，因为您设置为不自动提交错误消息。发生的事情是，这为客户端提供了从上次提交的偏移量重放的机会。

将 maxAttempts 设置为无穷大的问题是当发生不可恢复的错误时，您将让您的侦听器一遍又一遍地尝试使用消息。

更好的方法可能是为这些消息设置一个 dlq，然后使用 PollableChannel 定期轮询消息并尝试重新处理它们，这将为您的外部资源恢复提供一些时间。