如何使用Reactor/Mono将阻塞Java代码改造为响应式？

你当前用forEach + subscribe触发异步任务的写法，没法收集到目标结果列表——subscribe是异步触发执行，不会等待所有任务完成，也没有机制把每个getReportAndSave返回的Boolean汇总成列表。要实现和原阻塞代码等价的响应式非阻塞逻辑，得用响应式流的操作符来组合流程并收集结果。

核心修正思路 #

原阻塞代码的逻辑是：并行遍历siteIds → 调用异步API（generateReport）→ 阻塞等待API响应 → 保存报告并返回结果 → 收集所有结果到列表。响应式写法需要用Flux替代并行流，用操作符处理异步步骤，最后通过响应式的收集方法获取结果列表。

代码实现 #

1. 响应式环境推荐写法（无阻塞，返回Mono<List>） #

如果你的应用是响应式环境（比如Spring WebFlux），应该保持流程非阻塞，返回Mono<List<Boolean>>而非直接阻塞获取结果：

// 构建响应式流，并行处理siteIds
Mono<List<Boolean>> listResultMono = Flux.fromIterable(siteIds)
        .parallel() // 开启并行处理，对应原代码的parallel()流
        .runOn(Schedulers.boundedElastic()) // 指定并行执行的调度器，避免阻塞主线程
        .flatMap(siteId -> generateReport(authToken, requestParams, siteId)) // 处理异步API调用，展平Mono结果
        .map(baseResponse -> getReportAndSave(authToken, baseResponse.getRequestId())) // 执行保存操作，返回Boolean
        .sequential() // 将并行流转回串行，方便后续收集结果
        .collectList(); // 收集所有Boolean结果到List，返回Mono<List<Boolean>>

// 后续可通过subscribe处理结果（非阻塞）
listResultMono.subscribe(saveResults -> {
    // saveResults就是每个siteId的保存成功标识列表
    // 这里写结果处理逻辑
});

2. 兼容阻塞逻辑的写法（同步获取结果） #

如果必须像原代码一样同步拿到List<Boolean>（非响应式环境），可以在最后调用block()等待结果，但这会阻塞当前线程，仅建议在非响应式场景下使用：

List<Boolean> listResult = Flux.fromIterable(siteIds)
        .parallel()
        .runOn(Schedulers.boundedElastic())
        .flatMap(siteId -> generateReport(authToken, requestParams, siteId))
        .map(baseResponse -> getReportAndSave(authToken, baseResponse.getRequestId()))
        .sequential()
        .collectList()
        .block(Duration.ofMinutes(asyncCallTimeout)); // 阻塞等待所有任务完成

关键注意点 #

• 别用forEach+subscribe收集结果：subscribe仅用于触发流执行，无法同步获取汇总结果，异步场景下forEach结束时任务可能还在执行，根本拿不到完整列表。
• 阻塞操作线程隔离：如果getReportAndSave是阻塞式操作（比如操作Blob存储是同步阻塞的），需要把它包装到Mono.fromCallable()并指定调度器，避免阻塞响应式线程：

  .flatMap(baseResponse -> Mono.fromCallable(() -> 
          getReportAndSave(authToken, baseResponse.getRequestId()))
          .subscribeOn(Schedulers.boundedElastic()))
  

• 并行调度器选对：Schedulers.boundedElastic()适合处理I/O密集型阻塞操作，不会耗尽线程资源，比Schedulers.parallel()更安全。

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