Java Reactor Flux中flatMap抛出异常无法后续处理的问题求助

当在flatMap之后使用onErrorResume处理异常时，仅能捕获第一个异常，后续异常会触发onErrorDropped错误。这是因为Reactor流遵循单错误语义——一旦流发出错误信号，整个流就会终止，后续内部流的异常无法被下游处理，只能被丢弃。

下面提供两种可行的解决方案，实现flatMap之后统一处理所有内部流异常的需求：

方案1：用封装类统一包装结果与异常 #

通过给每个flatMap内部的流添加异常捕获，将成功结果和异常都封装为统一的对象，避免异常直接终止主流，之后在flatMap下游统一处理所有封装对象。

代码示例： #

// 定义结果封装类，区分成功值与异常
class Result<T> {
    private final T value;
    private final Throwable error;

    private Result(T value, Throwable error) {
        this.value = value;
        this.error = error;
    }

    public static <T> Result<T> success(T value) {
        return new Result<>(value, null);
    }

    public static <T> Result<T> error(Throwable error) {
        return new Result<>(null, error);
    }

    public T getValue() { return value; }
    public Throwable getError() { return error; }
    public boolean isSuccess() { return error == null; }
}

// 流处理逻辑
Flux.range(1, 3)
    .flatMap(i -> {
        Mono<Integer> innerMono;
        // 模拟部分内部流抛出异常
        if (i % 2 != 0) {
            innerMono = Mono.error(new Exception("ER错误"));
        } else {
            innerMono = Mono.just(i);
        }
        // 捕获内部流异常，封装为Result对象，避免终止主流
        return innerMono
            .map(Result::success)
            .onErrorResume(e -> Mono.just(Result.error(e)));
    })
    // flatMap之后统一处理所有结果与异常
    .doOnNext(result -> {
        if (!result.isSuccess()) {
            System.out.println("处理异常: " + result.getError().getMessage());
        } else {
            System.out.println("处理成功值: " + result.getValue());
        }
    })
    // 可选：过滤掉异常结果，仅保留成功值继续后续处理
    .filter(Result::isSuccess)
    .map(Result::getValue)
    .subscribe();

方案2：使用materialize/dematerialize包装流信号 #

利用materialize操作符将流的所有信号（onNext/onError/onComplete）转化为Signal对象，这样异常不会直接终止流，而是被包装为Signal元素，之后在下游处理这些Signal中的异常，最后可通过dematerialize还原正常的成功流。

代码示例： #

Flux.range(1, 3)
    .flatMap(i -> {
        // 模拟部分内部流抛出异常
        if (i % 2 != 0) {
            return Mono.error(new Exception("ER错误"));
        }
        return Mono.just(i);
    })
    // 将流的所有信号转化为Signal对象，异常不再直接终止流
    .materialize()
    // 统一处理所有信号
    .doOnNext(signal -> {
        if (signal.isOnError()) {
            System.out.println("处理异常: " + signal.getThrowable().getMessage());
        } else if (signal.isOnNext()) {
            System.out.println("处理成功值: " + signal.get());
        }
    })
    // 可选：还原为仅包含成功值的流，继续后续处理
    .dematerialize()
    .subscribe();

原方案失效的核心原因 #

Reactor遵循Reactive Streams规范，流具有单错误语义：一旦流发出onError信号，整个流立即终止，不再处理后续元素。当flatMap并发处理多个内部流时，第一个抛出的异常会触发主流的onError信号，触发onErrorResume后流终止，后续内部流的异常没有下游订阅者接收，因此会被丢弃并触发onErrorDropped警告。

内容的提问来源于stack exchange，提问作者Максим Цюпко