Reactor的FlatMap是否为异步操作？实操场景疑问

核心结论先明确 #

你看到的「未应用subscribeOn就同步逐个处理」的说法不准确。FlatMap的执行是同步还是异步，核心取决于内部流（即customerRepository.save()）本身是否为异步非阻塞操作，而非是否指定subscribeOn。

针对你的示例场景分析 #

假设itemDetailsCrudRepository.getItems()返回10个元素的Flux：

• 如果customerRepository.save()是异步非阻塞实现（比如基于R2DBC的响应式数据库操作）：
  • FlatMap会默认并行触发所有10个save请求（Reactor中FlatMap的默认并发上限是128，10远小于这个值），不会逐个等待前一个完成再执行下一个。
  • 此时不需要额外指定subscribeOn，内部的异步操作会自动在合适的线程上执行，数据流全程非阻塞。
• 如果customerRepository.save()是阻塞实现（比如传统JDBC同步操作）：
  • 哪怕用了FlatMap，也会逐个同步处理——因为前一个save操作会阻塞当前线程，导致后续元素的处理无法开始。
  • 这种情况下，你需要把阻塞的save操作放到专门的线程池（比如Schedulers.boundedElastic()），才能实现并行处理，示例代码调整如下：

    fun updateDetails() {
        itemDetailsCrudRepository.getItems()
            .flatMap { item ->
                customerRepository.save(someTransferObject.toEntity(item))
                    .subscribeOn(Schedulers.boundedElastic()) // 将阻塞操作转移到弹性线程池
            }
    }
    

纠正对subscribeOn的误解 #

subscribeOn(Schedulers.parallel())的作用是指定整个流的订阅阶段（包括上游getItems()的执行）运行在哪个线程池，它并不直接控制FlatMap内部流的并行性：

• 如果上游getItems()是阻塞操作，subscribeOn可以把它转移到非事件循环线程，避免阻塞主线程；
• 但如果内部的save()本身是异步的，有没有subscribeOn都不影响FlatMap并行触发save请求。

FlatMap vs Map的关键区别（补充你提到的基础疑问） #

• Map：同步转换每个元素，仅做元素的类型/值转换，不会生成新的流，全程在当前线程执行；
• FlatMap：为每个元素生成一个新的流（比如你的save()返回的Mono），然后把这些流的结果合并到最终的Flux中。合并的方式是并行还是串行，完全由内部流的异步性决定——内部流异步则并行，内部流阻塞则串行（除非手动指定线程池）。

额外注意点 #

• FlatMap默认的并发上限是128，你可以通过flatMap({...}, maxConcurrency)参数调整，比如flatMap({...}, 5)会限制同时最多处理5个元素；
• 如果需要严格保证元素的处理顺序，可以用concatMap代替FlatMap——concatMap会逐个处理元素，保证输出顺序，但牺牲了并行性。

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