Project Reactor timeout

Question

我正在一个项目反应堆车间工作，并且坚持以下任务：

/**
     * TODO 5
     * <p>
     * For each item call received in colors flux call the {@link #simulateRemoteCall} operation.
     * Timeout in case the {@link #simulateRemoteCall} does not return within 400 ms, but retry twice
     * If still no response then provide "default" as a return value
     */

我无法解决的问题是Flux实际上从未抛出TimeOutException！我可以在控制台日志中观察到这一点：

16:05:09.759 [main] INFO Part04HandlingErrors - Received red delaying for 300 
16:05:09.781 [main] INFO Part04HandlingErrors - Received black delaying for 500 
16:05:09.782 [main] INFO Part04HandlingErrors - Received tan delaying for 300 

尽管它似乎并没有改变行为，但我尝试按照陈述的顺序进行操作。注意：此外，我尝试了timeout（）的重载变体，如果没有发出任何元素，该变体接受应返回的默认值。

public Flux<String> timeOutWithRetry(Flux<String> colors) {

        return colors
                .timeout(Duration.ofMillis(400))
                //.timeout(Duration.ofMillis(400), Mono.just("default"))
                .retry(2)
                .flatMap(this::simulateRemoteCall)
                .onErrorReturn(TimeoutException.class, "default");

    }

有人可以弄清楚为什么不发生超时吗？我怀疑该机制不以某种方式“绑定”到flatMap调用的方法。

为了完整性：辅助方法：

public Mono<String> simulateRemoteCall(String input) {
        int delay = input.length() * 100;
        return Mono.just(input)
                .doOnNext(s -> log.info("Received {} delaying for {} ", s, delay))
                .map(i -> "processed " + i)
                .delayElement(Duration.of(delay, ChronoUnit.MILLIS));
    }

Answer 1

Flux上的超时观察后续元素之间经过的时间。但是，当您使用flatMap时，您可以立即使用并发性，元素之间的延迟实际上将为零（假设colors Flux由内存中的列表提供）。因此，不应将此运算符直接放在Flux上以实现您的目标。