使用 Spring Kafka 处理关于死信主题的消息

Question

我们目前使用 Spring Kafka（在 Spring Boot 应用程序中）进行了死信主题 (DLT) 配置。我们在SeekToCurrentErrorHandler 中使用DeadLetterPublishingRecoverer。我们将后一个分配给ConcurrentKafkaListenerContainerFactory。

在处理我们的第一条消息时；由于我们的服务中出现了一个愚蠢的错误，我们最终遇到了一些 NullPointerException 异常，并且 20% 的消息最终出现在 DLT 上（这是预期的行为，对我们来说是完美的）。

我们修复了这个错误，但现在我们想再次处理这 20% 的消息。我们看到的可能性：

• 编写一个将消息从 DLT 复制到原始主题的小应用程序
• 在从 DLT 读取的应用程序中添加第二个 @KafkaEventListener

解决方案 2 是我首选的解决方案，因为将其移回原始主题也意味着其他消费者组会再次收到消息（通常应该没问题，因为我们所有的服务都是幂等的）。

我想知道是否有其他最佳实践来解决这个问题。 如果没有，我还想知道如何为 DLT 动态激活/停用 @KafkaEventListener（因为您不想一直使用此侦听器）

感谢您的反馈！

乔辰

Answer 1

解决方案 2 对我来说很完美。

我还想知道如何动态激活/停用 DLT 的@KafkaEventListener（因为你不想拥有这个 一直在听）

您可以使用自 2.2 以来引入的 @KafkaListener 属性 autoStartup。

@Autowired
private KafkaListenerEndpointRegistry registry;

@KafkaListener(id = "123", topics = "XXX.DLT", autoStartup = "true"){
    //do your processing
}

//After you are done
registry.getListenerContainer("123").stop();