如何重试来自 kafka 的失败消息？

Question

我的spring-boot application(consumer)进程消息来自Apache Kafka. 周期性地，按摩无法处理并且消费者抛出异常。无论如何，消费者提交抵消。我可以区分 Kafka 中的成功消息和失败消息吗？我想，我不能。这是真的吗？如果是真的，我有一个主要问题：

如何重试失败消息？我知道一些方法，但我不确定它们的正确性。

1) 将偏移量更改为早期。但通过这种方式，成功消息也会重试。

2）当我捕获异常时，我将此消息发送到另一个主题（例如错误主题）。但看起来很难。

3）别的东西（你的变体）

Answer 1

如果你想要至少一次保证，一般模式如下：

• 禁用自动提交（设置enable.auto.commit为 false）
• 消费消息

• 对于每条消息：

  • 如果没有错误，则提交偏移量
  • 如果出错，请根据需要重试多次
  • 如果成功，提交
  • 如果你想放弃，记录或发布到错误队列（用于分析或稍后重试）

• 重复

Answer 2

使用SeekToCurrentErrorHandler。它将重新定位偏移量以重播消息（默认情况下为 10 次，但可配置）。

重试次数用尽后，它会调用可以执行某些操作的“恢复程序”，例如DeadLetterPublishingRecoverer。