Spring Cloud Stream Kafka DLQ 通过 Retryable/Recover

Question

我的用例是接收 Kafka 消息，在休息呼叫中多次尝试，并在用尽时将失败的消息转储到 Kafka DLQ 主题。

    @StreamListener(EventSource.SOME_CHANNEL)
    public void processMessage(Message<?> unsolicitedMessage) {
         String aString = .....   
         oneService.act(aString);      
    }

@Retryable 在处理多次尝试的逻辑方面工作得很好。

    @Retryable(value = {OneException.class, TwoException}, maxAttempts = 3, 
    backoff = @Backoff(delay = 1000))
    public boolean act(String message, String endPoint) {
         //do stuff
    }

为了让 Spring Cloud Stream 内置的 Kafka DLQ 发布启动 (enableDlq: true)，异常需要冒泡到 @StreamListener 注释的方法，以便 Kafka binder 执行必要的操作。

但是，在这样做时，我无法利用 @Recover 带注释的方法，该方法在重试后完美地降落：

@Recover
public boolean recoverOnToDLQ(OneException ex, String message, String 
endPoint) {
    throw ex; //Required for StreamListener Kafka DLQ to kick in! 
}

问题：有没有办法从@Recover 方法中触发Kakfa DLQ 发布而不重新抛出异常？

因为如果我只用它来重新抛出，我相信我不会有效地利用其中获得的更严格的控制。这也将简化单元测试用例，并更好地捕获代码级别的逻辑？关于如何更好地处理这个问题有什么想法吗？

截至目前，我使用的是 spring-cloud、spring-cloud-stream 和 spring-retry 的所有最新版本。

Answer 1

可以做到，但问题是“你为什么要这样做？”。

活页夹已内置重试；只需将异常扔回绑定器，它会在重试用尽后将数据发送到 DLQ。使用 binder 消费者重试属性配置重试。您不需要一行额外的代码。

通过使用 @Retryable，您将嵌套 2 个 RetryTemplates（除非您通过将使用者 maxAttempts 属性设置为 1 来禁用活页夹重试）。

当然，您可以配置自己的 DLQ 目的地，然后在您的恢复器中简单地写入您想要的任何内容。但是，要使用活页夹的内置 DLQ 发布者，您必须制作一个特殊的 ErrorMessage（具有发布者需要的属性）并将其发送到绑定的错误通道。发布者需要原始 kafka ConsumerRecord，您必须重新创建它，因为它对侦听器不可用。

总而言之，对于您的用例，仅使用活页夹的重试配置似乎要简单得多。

编辑

在 2.0.x 中，您可以将 RetryTemplate @Bean 添加到您的应用程序中，它将用于所有消费者绑定（覆盖绑定属性）。

可以使用任何重试策略、可重试异常等自定义此模板。

在 2.0.2 中，必须使用 @StreamRetryTemplate 进行限定；这是一个修复，因为 any RetryTemplate 会覆盖属性，这可能不是所需的行为。