4

我正在使用 reactor kafka 发送 kafka 消息并接收和处理它们。在接收 kakfa 有效负载时，我会进行一些反序列化，如果出现异常，我只想记录该有效负载（通过保存到 mongo ），然后继续接收其他有效负载。

为此，我使用以下方法 -

@EventListener(ApplicationStartedEvent.class)
public void kafkaReceiving() {
   for(Flux<ReceiverRecord<String, Object>> flux: kafkaService.getFluxReceives()) {
       flux.delayUntil(//some function to do something)
       .doOnNext(r -> r.receiverOffset().acknowledge())
       .onErrorResume(this::handleException()) // here I'll just save to mongo 
       .subscribe();
   }
}


private Publisher<? extends ReceiverRecord<String,Object>> handleException(object ex) {
 // save to mongo
 return Flux.empty();
}

在这里，我希望每当我在接收有效负载时遇到异常，onErrorResume 应该捕获它并记录到 mongo，然后当我发送到 kafka 队列时我应该继续接收更多消息。但是，我看到异常发生后，即使调用了 onErrorResume 方法，但我无法再处理发送到 Kakfa 主题的消息。有什么我可能在这里遗漏的吗？

4

2 回答 2

3

如果您需要优雅地处理错误，您可以在onErrorResume里面添加delayUntil：

flux
    .delayUntil(r -> {
        return process(r)
            .onErrorReturn(e -> saveToMongo(r));
    });
    .doOnNext(r -> r.receiverOffset().acknowledge())
    .subscribe();

反应式操作符将错误视为终止信号，并且，如果您的内部逻辑（内部delayUntil）抛出错误，delayUntil将终止序列，并且onErrorReturn 之后 delayUntil不会使其继续处理来自 Kafka 的事件。

于 2020-07-17T10:16:14.520 回答
0

正如@bsideup 所提到的，我最终没有从反序列化器中抛出异常，因为 kafka 无法为该记录提交偏移量，并且没有干净的方法可以忽略该记录并继续进一步消费记录因为我们没有记录的偏移信息（因为它格式错误）。因此，即使我尝试使用反应式错误运算符忽略记录，轮询也会获取相同的记录，然后消费者就会陷入困境

于 2020-08-21T17:13:55.773 回答