Reactor中compose与flatMap的差异疑问：二者表现为何一致？

嘿，这个问题问得特别到位！很多刚上手Reactor的开发者都会在compose和flatMap上打迷糊，毕竟在你给出的简单场景里，二者的输出看起来完全一致，但它们的设计初衷和底层行为其实有本质区别，咱们掰开揉碎了说：

核心定位：一个管序列，一个管元素 #

• compose是序列级的操作符：它接收的是整个上游Flux/Mono对象，作用是对整个序列的执行逻辑做“批量改造”，属于Reactor里的「元操作符」。它只会在序列被订阅时初始化一次操作链，后续所有元素都复用这套逻辑。
• flatMap是元素级的操作符：它针对上游发射的每一个元素，都会单独创建一个新的Flux/Mono子序列，然后把所有子序列的元素合并到下游输出。简单说，上游有多少个元素，它就会执行多少次内部的子序列创建逻辑。

用你的测试代码改改，差异一眼可见 #

先把你给出的测试代码格式化好，方便对比：

你的compose测试代码 #

@Test 
public void compose() throws InterruptedException { 
    Scheduler mainThread = Schedulers.single(); 
    Flux.just(("old element")) 
        .compose(element -> Flux.just("new element in new thread") 
            .subscribeOn(mainThread) 
            .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName()))) 
        .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName())) 
        .subscribe(System.out::println); 
    Thread.sleep(1000); 
}

你的flatMap测试代码 #

@Test 
public void flatMapVsCompose() throws InterruptedException { 
    Scheduler mainThread = Schedulers.single(); 
    Flux.just(("old element")) 
        .flatMap(element -> Flux.just("new element in new thread") 
            .subscribeOn(mainThread) 
            .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName()))) 
        .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName())) 
        .subscribe(System.out::println); 
    Thread.sleep(1000); 
}

现在咱们修改上游，让它发射3个元素，再加个初始化日志，就能看到差异了：

差异演示代码 #

@Test
public void composeVsFlatMapDifference() throws InterruptedException {
    Scheduler mainThread = Schedulers.single();
    
    // 测试compose：序列级操作，只初始化一次
    Flux.just("elem1", "elem2", "elem3")
        .compose(flux -> {
            System.out.println("👉 Compose: 初始化全局操作链");
            return flux.map(e -> "compose处理后的-" + e)
                .subscribeOn(mainThread);
        })
        .subscribe(System.out::println);

    Thread.sleep(500);
    System.out.println("--- 分割线 ---");

    // 测试flatMap：元素级操作，每个元素都触发一次子序列创建
    Flux.just("elem1", "elem2", "elem3")
        .flatMap(e -> {
            System.out.println("👉 FlatMap: 为元素[" + e + "]创建子序列");
            return Flux.just("flatMap处理后的-" + e)
                .subscribeOn(mainThread);
        })
        .subscribe(System.out::println);

    Thread.sleep(1000);
}

执行这段代码，输出会是：

👉 Compose: 初始化全局操作链
compose处理后的-elem1
compose处理后的-elem2
compose处理后的-elem3
--- 分割线 ---
👉 FlatMap: 为元素[elem1]创建子序列
👉 FlatMap: 为元素[elem2]创建子序列
👉 FlatMap: 为元素[elem3]创建子序列
flatMap处理后的-elem1
flatMap处理后的-elem2
flatMap处理后的-elem3

适用场景总结 #

• 如果你需要给整个序列套一套通用逻辑（比如统一加日志、统一指定调度器、封装复用的操作链），选compose，效率更高，因为只初始化一次。
• 如果你需要针对每个元素做独立的异步处理（比如每个元素调用不同的HTTP接口、每个元素生成不同的子数据流），选flatMap，它天然支持多元素的并行/异步处理。

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