Spring Reactor Flux，如何订阅和稍后阻塞直到全部完成答案

【问题标题】：Spring Reactor Flux, how to subscribe and later block until all doneSpring Reactor Flux，如何订阅和稍后阻塞直到全部完成
【发布时间】：2021-09-10 00:42:34
【问题描述】：

如果我想同步并阻塞直到所有元素都完成，Spring reactor 会提供 blocklast()

但是如果我想继续一点然后阻塞直到所有元素都完成怎么办？

（我不想使用 isDisposed 执行繁忙的等待）

我需要自己用 onComplete 触发的信号来做，还是有更好的内置 API？

//reactor provides blocklast if I want to be synchronous and block until all elements are done
//Integer data = Flux.range(1,10).delayElements(Duration.ofSeconds(1)).doOnNext((Integer next) -> System.out.println(next + " on thread " + Thread.currentThread().getName())).blockLast();
 
//but what if I want to continue a bit and then block
//do I need to do it myself like this? is there a better way?
Object signal = new Object(); 
Flux.range(1,10).delayElements(Duration.ofSeconds(1)).doOnNext((Integer next) -> System.out.println(next + " on thread " + Thread.currentThread().getName())).doOnComplete(()->{synchronized(signal) { signal.notify();}}).subscribe();
// do some other work here, then wait for done
synchronized (signal) {
    signal.wait();
}
System.out.println("All done");

【问题讨论】：

标签： java spring reactive reactor

【解决方案1】：

Mono.zip(
                        Mono.fromCallable(() -> {
                            //something
                            System.out.println("Completed doing");
                            return 1;
                        }).delaySubscription(Duration.ofSeconds(12)).subscribeOn(Schedulers.boundedElastic()),
                        Flux.range(1, 10)
                                .delayElements(Duration.ofSeconds(1))
                                .doOnNext((Integer next) -> System.out.println(next + " on thread " + Thread.currentThread().getName()))
                                .last()
                )
                .subscribe(__ -> System.out.println("All Done"));

【讨论】：

不，这不能回答我的问题。我想先做“某事”，而不是等待完成，然后阻塞直到全部完成。在您的解决方案中，“某事”只会在所有数据完成后才会发生。
你可以Mono.zip。更新了答案

【解决方案2】：

正如我在您的问题中看到的那样，您想要一些其他 api 来替换 blockLast，对吗？据我所知，您必须自己实施。没有为此目的的内置 api。

【讨论】：