FlatMapDelayError错误传播不一致问题咨询

核心差异原因 #

你观察到的两种行为差异，本质是错误来源的不同导致Reactor的处理逻辑完全区分：

• 同步抛出RuntimeException：这是在flatMapDelayError的映射函数执行过程中同步发生的异常，属于“处理上游onNext信号时的执行错误”，而非子Flux发出的错误。Reactor会将这种异常视为终端错误，立即触发上游订阅的取消，同时将异常向下传播，因此StepVerifier能正常捕获错误并完成验证。
• 返回Flux.error(RuntimeException)：这是创建了一个主动发出onError信号的子Flux，属于子流的异步错误。flatMapDelayError的核心设计就是延迟子流错误的传播——它会缓存子流的错误，继续请求上游的后续元素，直到上游所有元素处理完毕，才会将收集到的所有错误合并为CompositeException抛出。这也是为什么你的第二个测试会“挂起”：它需要等上游的4个元素全部处理完成后，才会输出错误信号。

使用flatMapDelayError的核心注意点 #

1. 明确设计目标：flatMapDelayError的存在意义就是为了在子流出现错误时，不立即终止整个流，而是继续处理其他元素，直到所有子流完成。如果你希望错误立即触发上游取消，就不应该使用这个操作符，改用普通的flatMap即可。
2. 区分同步/异步错误：映射函数的同步异常和子流的异步错误是两种完全不同的错误场景，Reactor对它们的处理逻辑天然不同，不要期望两者行为完全一致。如果需要统一行为，要么将子流错误转为同步抛出（比如在映射函数中捕获子流错误并同步抛出），要么放弃使用flatMapDelayError。
3. 错误合并的特性：当多个子流都产生错误时，flatMapDelayError会将所有错误合并成CompositeException抛出，而非只抛出第一个错误。这一点在编写错误处理逻辑时需要注意。

验证修正示例 #

如果你希望第二个测试的行为和第一个一致，可以将子流错误转为同步抛出：

Flux<Object> propagatedFlux = Flux.just(0, 1, 2, 3).log()
                                  .flatMapDelayError(integer -> {
                                      throw new RuntimeException(); // 和第一种方式统一行为
                                  }, 1, 1);

或者如果你需要保留子流错误但希望立即终止，改用普通flatMap：

Flux<Object> propagatedFlux = Flux.just(0, 1, 2, 3).log()
                                  .flatMap(integer -> Flux.error(new RuntimeException()));

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