Project Reactor中Mono链subscribeOn用法及非阻塞实现正确性咨询

一、你的非阻塞实现是否正确？ #

不正确，存在两个关键问题：

1. 阻塞任务被重复执行 #

Mono<A> a和Mono<B> b是「冷序列」——每次被订阅，都会重新执行对应的getA()/getB()。你的代码里，c和d各自通过Mono.zip(a,b,...)订阅了a和b各一次，导致getA()和getB()会被执行两次，这不仅浪费资源，还可能因为重复执行耗时任务导致结果不一致（如果方法有副作用的话），和原阻塞代码的逻辑不符。

2. 阻塞任务可能阻塞关键线程 #

你直接在Mono.zip的合并函数里调用getC()/getD()，这两个方法是耗时阻塞的，会直接阻塞当前执行合并逻辑的线程。如果这个线程是线程池里的核心线程，会降低整体并发能力。

修正后的实现示例 #

public E someMethodUnblocking() {
    // 缓存a、b的结果，避免重复执行耗时任务
    Mono<A> a = Mono.fromCallable(this::getA)
                     .subscribeOn(Schedulers.boundedElastic())
                     .cache();
    Mono<B> b = Mono.fromCallable(this::getB)
                     .subscribeOn(Schedulers.boundedElastic())
                     .cache();

    // 将getC、getD也包装为异步Mono，指定单独的阻塞线程池执行
    Mono<C> c = Mono.zip(a, b)
                    .flatMap(tuple -> 
                        Mono.fromCallable(() -> getC(tuple.getT1(), tuple.getT2()))
                            .subscribeOn(Schedulers.boundedElastic())
                    );
    Mono<D> d = Mono.zip(a, b)
                    .flatMap(tuple -> 
                        Mono.fromCallable(() -> getD(tuple.getT1(), tuple.getT2()))
                            .subscribeOn(Schedulers.boundedElastic())
                    );

    // 最后合并c、d并阻塞获取结果
    return Mono.zip(c, d, this::getE)
               .block();
}

二、移除Mono a和Mono b的subscribeOn()会产生什么差异？ #

核心差异在于**getA()/getB()的执行线程不再可控**，具体影响：

• 线程不确定性：没有subscribeOn时，getA()/getB()会在「首次订阅它们的线程」上执行。如果后续代码中a/b被不同的线程订阅，这两个耗时任务就可能跑到任意线程上。
• 可能阻塞主线程：如果最终的block()是在主线程调用的，那么getA()/getB()会直接在主线程执行，导致主线程被阻塞——这完全违背了你改成非阻塞实现的初衷。
• 依赖下游线程不可靠：即使你在c/d上添加了subscribeOn，虽然getA()/getB()可能会在boundedElastic线程执行，但这种依赖下游配置的方式非常脆弱，一旦后续代码修改了c/d的线程配置，或者直接订阅a/b，就会引发阻塞问题。

简单说：移除a/b的subscribeOn，等于放弃了对这两个耗时任务执行线程的控制权，很容易导致阻塞关键线程，破坏非阻塞的设计。

内容的提问来源于stack exchange，提问作者Mel H