Reactive MongoDB中Flux数据流保存未持久化问题排查：并行执行保存与删除操作的订阅疑惑

你踩了响应式编程里一个非常常见的坑——冷流的订阅触发规则，先直接说核心问题：doOnNext根本不会触发订阅，它只是一个副作用操作，你的保存管道因为没有被订阅，所以完全没执行！

为什么之前的代码没生效？ #

先看你的保存代码：

tweetRepository.saveAll(tweetsByUserId).collectList().map(lis -> lis.size()).doOnNext(System.out::println);

这段代码只是构建了响应式管道的逻辑，但没有任何订阅行为。响应式流（比如Flux/Mono）是「冷流」——只有当有订阅者（subscriber）主动订阅时，整个管道才会开始发射数据、执行操作。

doOnNext是中间操作（intermediate operator），它的作用是在数据流经过时执行副作用（比如打印日志），但它不会触发订阅。而collectList()虽然把Flux转换成了Mono，但这个Mono依然需要被订阅才会执行上游的saveAll操作。

反观你的删除操作deleteTweets(tweetsByUserId)，它能生效的原因大概率是：这个方法内部的管道被订阅了，或者它返回的Mono/Void被上层代码（比如Spring WebFlux控制器）订阅了，所以整个删除流程启动了。

正确的并行执行方式 #

要让保存和删除两个操作并行执行且都被触发，你需要把两个操作的管道合并，确保它们都被订阅。推荐用Mono.when（适合无返回值或只关心完成状态）或者Flux.zip（需要合并结果）来实现：

// 1. 定义保存操作的管道，返回保存的数量
Mono<Integer> saveTweetCount = tweetRepository.saveAll(tweetsByUserId)
    .collectList()
    .map(List::size)
    .doOnNext(savedCount -> System.out.println("成功保存 " + savedCount + " 条推文"));

// 2. 定义删除操作的管道（假设deleteTweets返回Mono<Void>）
Mono<Void> deleteTweetResult = deleteTweets(tweetsByUserId);

// 3. 并行执行两个操作，等待都完成后返回
return Mono.when(saveTweetCount, deleteTweetResult);

关键知识点回顾 #

• 冷流特性：所有Flux/Mono都是冷流，没有订阅就不会执行任何操作。
• 操作符分类：中间操作（map/filter/doOnNext等）只负责转换数据流，不会触发订阅；只有终端操作（subscribe/block/collectList等，或者被其他已订阅的管道引用）才会启动整个流。
• 并行执行的正确姿势：不要单独构建未订阅的管道，而是用Mono.when/Flux.zip这类操作符把多个操作合并，让它们共享同一个订阅触发。

另外注意：不要在Spring WebFlux等响应式框架里手动调用subscribe()，这会脱离框架的订阅生命周期管理，应该让框架来处理订阅（比如返回Mono/Void给控制器，框架会自动订阅）。

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