Reactor中subscribe的Consumer内部报错如何处理可不取消订阅

问题核心原因 #

onErrorContinue 操作符的作用范围是它上游的操作符链产生的异常，你把抛出异常的业务逻辑写在了subscribe的消费者函数中，这部分属于订阅终端的逻辑，不在onErrorContinue的捕获范围内，所以异常会直接触发订阅的错误回调，终止整个订阅流程。

两种可行解决方案 #

方案1：把业务逻辑移到上游操作符链中，让onErrorContinue可以捕获异常 #

把计算逻辑从subscribe回调移到上游的doOnNext/map等操作符内，就可以被onErrorContinue捕获处理：

public static void main(String[] args) {
    Flux.just(1,2,3)
            .doOnNext(i -> System.out.println(10/(2-i)))
            .onErrorContinue((err,i)-> System.out.println(err+" "+i))
            .subscribe();
}

运行输出：

10
java.lang.ArithmeticException: / by zero 2
-10

可以看到元素2触发的异常被捕获后，后续元素3仍正常处理，订阅不会终止。

方案2：在subscribe的消费者内部直接做异常捕获 #

如果必须把逻辑写在subscribe的回调中，直接在内部加try-catch处理异常，不要把异常抛出到订阅框架层面即可：

public static void main(String[] args) {
    Flux.just(1,2,3)
            .subscribe(i->{
                try {
                    System.out.println(10/(2-i));
                } catch (Exception e) {
                    System.out.println(e+" "+i);
                }
            }, System.out::println);
}

运行效果和方案1完全一致。

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