Spring Reactive - 重用 Mono 值

Ada*_*lik 4 java spring project-reactor spring-webflux

我有一系列Mono使用flatMap. 我设法将我的生产代码简化为这个测试用例:

@Test
public void test() {
    AtomicInteger iCounter = new AtomicInteger(1);
    Mono<String> iValueMono = Mono.fromSupplier(() -> {
        int iValue = iCounter.getAndIncrement();
        System.out.println("iValueMono CALL: " + iValue);
        return String.valueOf(iValue);
    });

    Mono<String> resultMono = Mono.just("X")
            .flatMap(append(iValueMono))
            .flatMap(append(iValueMono));

    StepVerifier.create(resultMono)
            .consumeNextWith(result -> assertThat(result).isEqualTo("X11"))
            .expectComplete()
            .verify();
}

private Function<String, Mono<String>> append(Mono<String> sMono) {
    return s -> sMono.map(v -> s + v);
}
Run Code Online (Sandbox Code Playgroud)

这打印:

iValueMono CALL: 1
iValueMono CALL: 2

org.junit.ComparisonFailure: 
Expected :"X11"
Actual   :"X12"
Run Code Online (Sandbox Code Playgroud)

我想 - 我现在明白这是不正确的 - 每次我iValueMonoappend()调用中映射 时,都会重新执行供应商以产生一个新值。我无法在生产代码中更改iValueMono实现方式(例如,使其有状态以存储值)。我如何实现这一点,以便只调用一次值供应商并获得最终结果“X11”?

当然,我对一种非阻塞、反应式的方式来做到这一点很感兴趣。

Ada*_*lik 7

使用Mono.cache()是答案:

将此 Mono 转换为热源并缓存最后发出的信号以供进一步的订阅者使用。

使用它:

Mono<String> iValueMono = Mono.fromSupplier(() -> {
    int iValue = iCounter.getAndIncrement();
    System.out.println("iValueMono CALL: " + iValue);
    return String.valueOf(iValue);
}).cache();
Run Code Online (Sandbox Code Playgroud)

只提供一次调用供应商的预期结果。