超时后停止从 KafkaReceiver 消费答案

【问题标题】：Stop consuming from KafkaReceiver after a timeout超时后停止从 KafkaReceiver 消费
【发布时间】：2021-06-01 22:24:55
【问题描述】：

我有一个通用的休息控制器：

    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？

【问题讨论】：

标签： spring-webflux project-reactor reactive-kafka

【解决方案1】：

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

相反，只需使用 overloaded timeout method 提供一个空的默认发布者以回退到：

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

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

【讨论】：