超时后停止从KafkaReceiver消费

Question

我有一个普通的rest控制器:

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 ?

Answer 1

timeout()方法有多个重载-您使用的是标准的在超时时抛出异常的方法。

相反，只需使用重载的timeout方法来提供一个空的默认发布者以回退到:

timeout(Duration.ofSeconds(2), Mono.empty())

(注意，在一般情况下，您可以显式捕获TimeoutException并使用onErrorResume(TimeoutException.class, e -> Mono.empty())回退到一个空的发布者，但这比在可能的情况下使用上述选项要好得多。)