如何解决Mono.fromCompletionStage外部调用抛异常时错误算子不生效问题

根因分析 #

你遇到的问题本质是外部方法调用时机早于Mono实例创建，错误处理算子仅能捕获响应流生命周期内的异常，无法捕获Mono构建阶段抛出的异常：
你写的Mono.fromCompletionStage(externalCall(i))代码中，externalCall(i)是立即执行的，在flatMap收到元素后会先调用该方法，再用返回的CompletionStage构造Mono实例。如果调用externalCall直接抛出异常，此时Mono还未完成创建，挂载在它上面的doOnError、onErrorResume等算子根本没有生效的机会，异常会直接向上游抛出，最终被全局错误捕获器处理，就是你看到的ErrorCallbackNotImplemented日志。

解决方案 #

方案1：使用Mono.defer延迟执行（适配所有Reactor版本） #

将externalCall的调用包裹在Mono.defer的Supplier函数中，将调用时机延迟到Mono被订阅时，此时抛出的异常会被Reactor自动纳入响应流错误处理链路：

private void example() {
    Flux.range(1, 2)
            .flatMap(i ->
                    // 用defer延迟externalCall的执行
                    Mono.defer(() -> Mono.fromCompletionStage(externalCall(i)))
                            .doOnNext(ni -> System.out.println("onNext: " + ni))
                            .doOnError(err -> System.err.println("onError: " + err.getMessage()))
                            .onErrorResume(e -> Mono.empty())
            )
            .subscribe();

    try {
        Thread.sleep(10000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

方案2：使用fromCompletionStageSupplier重载（Reactor 3.4.0及以上版本可用） #

Reactor提供了接受Supplier参数的Mono.fromCompletionStage重载，会自动延迟CompletionStage的生成并捕获调用过程的异常，代码更简洁：

.flatMap(i ->
        // 直接传入Supplier，无需手动包裹defer
        Mono.fromCompletionStage(() -> externalCall(i))
                .doOnNext(ni -> System.out.println("onNext: " + ni))
                .doOnError(err -> System.err.println("onError: " + err.getMessage()))
                .onErrorResume(e -> Mono.empty())
)

效果说明 #

修改后externalCall抛出的异常会被错误处理算子正常捕获，onErrorResume返回Mono.empty后，当前元素会被忽略，流会继续处理后续的元素，满足你忽略异常继续执行的需求。

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