超时后停止从 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()) 并回退到空发布者，但这比尽可能使用上述选项要好得多。）