3

我正在尝试使用 reactor-kafka 来消费消息。其他一切正常，但我想为失败的消息添加重试（2）。spring-kafka 默认已经重试了 3 次失败记录，我想使用 reactor-kafka 来实现同样的效果。

我正在使用 spring-kafka 作为响应式 kafka 的包装器。以下是我的消费者模板：

reactiveKafkaConsumerTemplate
                .receiveAutoAck()
                .map(ConsumerRecord::value)
                .flatMap(this::consumeWithRetry)
                .onErrorContinue((error, value)->log.error("something bad happened while consuming : {}", error.getMessage()))
                .retryWhen(Retry.backoff(30, Duration.of(10, ChronoUnit.SECONDS)))
                .subscribe();

让我们考虑一下consume方法如下

public Mono<Void> consume(MessageRecord message){
   return Mono.error(new RuntimeException("test retry"); //sample error scenario
}

我正在使用以下逻辑在失败时重试使用方法。

public Mono<Void> consumeWithRetry(MessageRecord message){
   return consume(message)
          .retry(2);
}

如果当前消费者记录因异常而失败，我想重试使用该消息。我试图用另一个 retry(3) 包装消费方法，但这并没有达到目的。最后一次 retryWhen 仅用于在 kafka rebalances 上重试订阅。

@simon-baslé @gary-russell

4

1 回答 1

0

以前在重试时我使用以下方法：

public Mono<Void> consumeWithRetry(MessageRecord message){
   return consume(message)
          .retry(2);
}

但它没有重试。添加 Mono.defer 后，上面的代码工作并添加了所需的重试。

public Mono<Void> consumeWithRetry(MessageRecord message){
   return Mono.defer(()->consume(message))
          .retry(2);
}
于 2021-10-01T09:40:38.347 回答