如何解决超时后停止从 KafkaReceiver 消费
private final KafkaReceiver<String,Domain> receiver;
@GetMapping(produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<Domain> produceFluxMessages() {
return receiver.receive().map(ConsumerRecord::value)
.timeout(Duration.ofSeconds(2));
}
我想要实现的是在一段时间内从 Kafka 主题中收集消息,然后停止消费并认为此通量已完成。如果我删除超时并在浏览器中打开它,我将永远收到消息,下载永远不会停止。并且这个超时消耗在 2 秒后停止,但我得到一个例外:
java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 2000ms in 'map' (and no fallback has been configured)
有没有办法在超时后成功完成 Flux?
解决方法
timeout()
方法有多个重载 - 您使用的是在超时时引发异常的标准方法。
相反,只需使用 overloaded timeout method 提供一个空的默认发布者以回退到:
timeout(Duration.ofSeconds(2),Mono.empty())
(请注意,在一般情况下,您可以使用 TimeoutException
显式捕获 onErrorResume(TimeoutException.class,e -> Mono.empty())
并回退到空发布者,但这比尽可能使用上述选项要好得多。)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。