将定期操作绑定到 Mono 生命周期的最佳方法是什么？

Question

我想在 Mono 处于活动状态时每秒执行一次操作。最好的方法是什么？

这是一个可行的选项，但似乎是一种解决方法：

// Some long life async action
Mono<String> asyncAction = Mono.delay(Duration.ofSeconds(60)).map(d -> "Hello");

Mono<String> periodicAction = Flux.interval(Duration.ofSeconds(1))
        .doOnNext(d -> {
            // Do something every second while the async action is running
        })
        .last()
        .flatMap(d -> Mono.never());

Mono.firstWithSignal(asyncAction, periodicAction)
        // Another logic
        .subscribe();

Answer 1

请考虑使用方法zip，例如：

static <T1,T2> Flux<Tuple2<T1,T2>>  zip(Publisher<? extends T1> source1, Publisher<? extends T2> source2)

https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#zip-org.reactivestreams.Publisher-org.reactivestreams.Publisher-

例子：

        Flux.zip(
                Flux.interval(Duration.ofSeconds(1)).doOnNext(System.out::println).log(),
                Mono.delay(Duration.ofSeconds(5)).map(d -> "Hello").log()
        )
                .map(Tuple2::getT2)
                .log()
                .as(StepVerifier::create)
                .expectNextCount(1)
                .verifyComplete()

在 zip 方法 java 文档中，我们可以读到：

将两个源压缩在一起，即等待所有源发出一个元素，然后将这些元素组合成一个{@link Tuple2}。运营商将继续这样做，直到任何来源完成。