详解Reactor如何优雅Exception异常处理

Reactor Exception异常处理

初识响应式编程的时候，除了从命令式的思维方式转变为函数式的编程方式外，其中有一个很大的不适应的地方就是在面对异常时该怎么处理，尤其是面对检查异常（Checked Exception)时更是不知所措。在遇到异常时，我们通用的处理方式就是打日志、降级兜底、重试三板斧，本文通过Project Reactor的文档以及源码来深入解读，在reactor中是如何优雅地实现这异常处理三板斧。

在介绍怎么使用前，我们先回顾下在用reactor编程的时候，遇到的几个问题：

• 遇到异常时，如果能处理，我该怎么兜底/降级
• 遇到无法处理的异常时，我该怎么打印日志，并往外抛
• 遇到声明了检查异常的方法时，该怎么处理
• 如果调用失败了（如请求超时），该如何重试
• 如果出现异常了，流里面的后续数据还会继续发送吗

异常处理的底层机制

在回答这些问题，就需要我们首先对reactor处理异常的机制要有理解。先说结论，如文档上说的：

Before you learn about error-handling operators, you must keep in mind that any error in a reactive sequence is a terminal event. Even if an error-handling operator is used, it does not let the original sequence continue. Rather, it converts the onError signal into the start of a new sequence (the fallback one). In other words, it replaces the terminated sequence upstream of it.

即，一旦出现了异常，那原先的数据流就会直接结束了，是没有办法再恢复的。所以如果要降级兜底，那只能再替换一个新的流。或者重试，但其实也是相当于创建了新流，只是数据和原先的一样。

那为什么原先的流就结束了呢？ 或者说怎么就结束了呢?

先抛开异常处理的话题，我们回到最基础的层面，如果要主动结束一个流，该怎么结束呢？比如一个网络连接，如果出现异常了，流结束了，该怎么释放资源呢？ 我们知道，每次订阅调用的时候，都会返回一个Disposable对象，如Disposable disposable = Flux.just(1,2,3).subscribe()。所以，如果要主动结束一个流，其实就是调用Disposable对象的dispose方法。再深入下去，就会发现，其dispose方法内部其实调用的是由publisher产生的subscription的cancel方法。只有调用cancel方法，才能完美的结束publisher并释放资源。所以，要想结束一个流，只有调用subscription的cancel方法。

所以，当出现异常时，原先的流会结束的原因，其实就是调用了subscription的cancel方法了。那是何时调用的呢，我们以FluxMap为例，看下源码。

public void onNext(T t) {
    if (done) {
        Operators.onNextDropped(t, actual.currentContext());
        return;
    }
    R v;
    try {
        v = Objects.requireNonNull(mapper.apply(t),
                "The mapper returned a null value.");
    }
    catch (Throwable e) {
        Throwable e_ = Operators.onNextError(t, e, actual.currentContext(), s);
        if (e_ != null) {
            onError(e_);
        }
        else {
            s.request(1);
        }
        return;
    }
    actual.onNext(v);
}

可以看到，在数据处理onNext()的方法内部，通常都有类似的try-catch结构。当出现异常时，会先对异常进行处理，确定是否需要处理异常以及怎么处理，其逻辑都在Operators.onNextError这个方法里。若要处理异常时，则会进入onError的流程里。
下面重点来看看Operators.onNextError这个方法，它主要主要包含了两件事：

• 这个异常要不要吃掉，当做非异常处理：具体可以参见Operators.onNextErrorStrategy方法
• 这个异常要不要往下传，即如果是严重异常时，则直接抛出；否则，对订阅进行cancel。详见Operators.onOperatorError方法.

减少篇幅起见，这里只贴下Operators.onOperatorError的方法，其实onNextErrorStrategy方法也很重要，它会从context里拿出对指定异常能处理的strategy来进行处理，通过传入的value以及产生的error决定是否要抛出这个异常。

public static Throwable onOperatorError(@Nullable Subscription subscription,
        Throwable error,
        @Nullable Object dataSignal, Context context) {

    Exceptions.throwIfFatal(error);
    if(subscription != null) {
        subscription.cancel();
    }

    Throwable t = Exceptions.unwrap(error);
    BiFunction<? super Throwable, Object, ? extends Throwable> hook =
            context.getOrDefault(Hooks.KEY_ON_OPERATOR_ERROR, null);
    if (hook == null) {
        hook = Hooks.onOperatorErrorHook;
    }
    if (hook == null) {
        if (dataSignal != null) {
            if (dataSignal != t && dataSignal instanceof Throwable) {
                t = Exceptions.addSuppressed(t, (Throwable) dataSignal);
            }
            //do not wrap original value to avoid strong references
            /*else {
            }*/
        }
        return t;
    }
    return hook.apply(error, dataSignal);
}

从onOperatorError的源码里可以看到：

• Exceptions.throwIfFatal决定是否直接抛出异常还是进入onError流程。比如JVM的异常，例如OutOfMemory之类的，就会直接抛出。
• 接下来，就是我们前面想要的答案，异常时数据流会结束的原因，就是默认会调用subscription.cancel()。
• 最后，是对异常处理的钩子hook，可以通过传入的onNext的数据dataSingal以及异常error进行处理，例如打印日志或者进行异常的转化，甚至是吃掉异常。

实操 —— try-catch-finally的平替

在了解了reactor中对异常处理的机制后，我们看看有哪些操作符可以用来代替以前命令式编程中的try-catch-finally的结构。遇到异常时，我们通用的方式就是打日志、降级兜底、重试三板斧，下面我们具体看看在reactor中是怎么实现的。

Flux.just(1,2,3)
    .doFinally(signalType -> System.out.println("Finally: [" + signalType + "]" ))
    .map(t -> {
        if (t == 2) {
          throw new IllegalArgumentException("Exception:"+t);
        }
        return t;
    })
    .doOnError(e -> System.out.println("log: error happened with [" + e.getMessage() + "]"))
    .doFinally(signalType -> System.out.println("Finally: [" + signalType + "]" ))
    .onErrorReturn(42)
//        .onErrorResume(e -> Flux.just(11,12,13))
    .subscribe(d -> System.out.println("data:" + d), e -> System.out.println("ERROR:" + e.getMessage()), () -> System.out.println("Completed!"));

这是一个使用示例，其输出如下，下面会对这些操作符进行介绍。

data:1
Finally: [cancel]
log: error happened with [Exception:2]
data:42
Completed!
Finally: [onError]

1. 降级兜底 - onErrorReturn/onErrorResume

当遇到异常的时候，可以使用onErrorReturn来处理，返回一个默认值，其底层实现其实用的还是onErrorResume。相比onErrorReturn只能返回一个默认值而言，onErrorResume更灵活，它可以根据不同的error类型，还实现不同的返回值，其fallback函数的入参是异常类型，返回的则是一个Publisher。所以从返回值也可以看出来，onErrorReturn/onErrorResume返回的是一个新的流，旧的流已经在发生异常的时候就结束了。

public final Flux<T> onErrorResume(Predicate<? super Throwable> predicate,
        Function<? super Throwable, ? extends Publisher<? extends T>> fallback) {
    Objects.requireNonNull(predicate, "predicate");
    return onErrorResume(e -> predicate.test(e) ? fallback.apply(e) : error(e));
}

这是onErrorResume的实现，其提供了特别灵活的处理方式，predicate决定是否要进行fallback，fallback下可以根据异常类型返回任意的数据流。假如不需要对异常fallback，即predicate为false时，则直接返回FluxError的封装，进入onError阶段。

再补充一点，既然fallback是返回的一个新的流，那么即可以fallback返回一个单值，例如onErrorReturn那样，也可以返回多个值的数据流，例如：

Flux.just(1,2,3)
    .map(t -> {
        if (t == 2) {
          throw new IllegalArgumentException("Exception:"+t);
        }
        return t;
    })
   .onErrorResume(e -> Flux.just(11,12,13))
   .subscribe(d -> System.out.println("data:" + d), e -> System.out.println("ERROR:" + e.getMessage()), () -> System.out.println("Completed!"));

输出结果会是：1 11 12 13。

其实，对于Flux而言，因为是多值的数据流，其很难根据异常error就进行合适的兜底，因为兜底往往取决于输入，而异常的时候往往丢失了数据data的信息了，所以对于Flux的降级，onErrorReturn/onErrorResume实用性不是太强，因为返回的新的流很难替代旧的流，甚至你都不知道旧的Flux流有多少数据量。

相比之下，onErrorReturn/onErrorResume用于单值的Mono就显得更为合适了。由于用法相同，这里就不过多赘述了。

所以，对于Flux流的降级兜底是个很困难的事情，有一种方式可以让onErrorReturn/Resume获取到当前的数据data，那就是利用前面小节中说到的，增加Operators.onOperatorError中的onOperatorErrorHook，把数据data塞入异常中再返回给onErrorReturn/Resume。

最后再看下FluxOnErrorResume是怎么实现降级的。

public void onError(Throwable t) {
    if (!second) {
        second = true;

        Publisher<? extends T> p;

        try {
            //nextFactory即fallback函数
            p = Objects.requireNonNull(nextFactory.apply(t),
            "The nextFactory returned a null Publisher");
        }
        catch (Throwable e) {
            Throwable _e = Operators.onOperatorError(e, actual.currentContext());
            _e = Exceptions.addSuppressed(_e, t);
            actual.onError(_e);
            return;
        }
        //重新订阅一个新的流，其source就是fallback函数产生的publisher
        p.subscribe(this);
    }
    else {
        actual.onError(t);
    }
}

当上游出现异常时，例如先前示例中的FluxMap，就会进入onError阶段，此时正好被onErrorResume的onError阶段拦截，然后利用fallback函数产生新的流，再重新订阅p.subscribe(this)。另外，也可以看出，新的流只会作用于onErrorResume之后的operator，前面的operator则不会有作用。

2. 打印日志 - doOnError

打印日志就比较简单了，可以用doOnError方法来实现。doOnError的底层则用的FluxPeek来实现，其作用是覆写了所有的接口，如onNext,onError, cancel等，通过覆写来实现hook。几乎所有doOnXXX的方法都是依赖FluxPeek实现的，例如log、doOnNext、doOnError等等。由于与本次的主题无关，不再赘述，感兴趣的可以自行翻看FluxPeek的实现。

需要注意的是：虽然doOnXXX主要用于打印日志，但如果doOnXXX内部出错，也会导致整个流结束，进入onError阶段。所以，也是有副作用的，仍然在主流程中。

3. finally - doFinally

try-catch-finally中的关键字finally可以通过方法doFinally来平替。需要注意的是doFinally方法的执行顺序以及触发时机。

通常，finally的含义是保证100%被执行，也就是出错onError的时候执行，正常结束onComplete也执行。但在reactor中，除了这两个事件外，还未能保证doFinally百分之百执行，还需要增加cancel的情况。其原因是，当出现异常后，对于异常的上游会走cancel流程，下游则走onError流程。如先前的示例，触发doFinally的信号分别是：cancel与onError。

最后再说一下执行顺序，如先前的示例中那样，doFinally并不是按出现的顺序执行，也不是一定是在最后执行的（这个区别与finally关键词差别很大）。其原因在于，当出现异常时，会先cancel掉原先的数据流，再调用onError处理（可以参见前面FluxMap的源码）。

所以，示例中，"Finally: [cancel]"会先被打印，然后才是onErrorReturn的执行，即进入onError阶段。

那为什么第二个doFinally虽然出现在onErrorReturn之前，但又是最后执行的呢？这是因为在实现doFinally的时候，先调用了下游的onError方法，再执行自身doFinally的方法，参见FluxDoFinally的实现：

public void onError(Throwable t) {
    try {
        actual.onError(t);
    }
    finally {
        runFinally(SignalType.ON_ERROR);
    }
}

这样就能符合try-catch-finally的执行顺序了。

所以，doFinally出现的位置很重要，若出现在异常前面，就会优先执行（不会像finally那样最后执行），若出现在异常后面，则会最后执行（类似finally）。

4. try-with-resource

对于try-with-resource，reactor也给了替代的实现，那就是using操作符:

public static <T, D> Flux<T> using(Callable<? extends D> resourceSupplier, Function<? super D, ? extends
        Publisher<? extends T>> sourceSupplier, Consumer<? super D> resourceCleanup, boolean eager) {
    return onAssembly(new FluxUsing<>(resourceSupplier,
            sourceSupplier,
            resourceCleanup,
            eager));
}

其中resourceSupplier是创建生成资源的函数，sourceSupplier则是针对生成的resource进行操作并产生数据流，resourceCleanup则是在结束后（不管成功还是失败）进行资源的释放。

以try-with-resource为例：

try (SomeAutoCloseable disposableInstance = new SomeAutoCloseable()) {
  return disposableInstance.toString();
}

利用using函数，则可以写成：

Flux.using(
        () -> new SomeAutoCloseable(), 
        disposableInstance -> Flux.just(disposableInstance.toString()), 
        AutoCloseable::close 
);

5. 重试 - retry / retryWhen

除了以上方式处理异常时，还有一种常见的方式就是重试。比如，我们调用某个接口超时时，通常会重试一次，这个时候可以使用retry方法，如：

Flux.interval(Duration.ofMillis(250))
    .map(input -> {
        if (input < 3) return "tick " + input;
        throw new RuntimeException("boom");
    })
    .retry(1)
    .subscribe(System.out::println, System.err::println); 

会对Flux流执行两次，其结果是：0 1 2 0 1 2，即当遇到data为3时，会重试一次。

其基本思想很简单，就是拦截onError流程，计算重试的次数，如果重试未超过，则重新订阅：

public void onError(Throwable t) {
    long r = remaining;
    if (r != Long.MAX_VALUE) {
        if (r == 0) {
            actual.onError(t);
            return;
        }
        remaining = r - 1;
    }
    resubscribe();
}

这里的remaining就是可以重试的次数，直到重试为0，再一次进入actual.onError。重新订阅的方法也很简单，就是把上游的source与下游的actual，再来一次subscribe：source.subscribe(actual)。

除了retry外，还有一个高级版本retryWhen，它除了能像retry那样重试固定的次数外，还能支持各种重试策略，由于retryWhen的源码相对复杂，这里不再叙述（毕竟本文不是源码解读），但除了重试策略有区别外，其重试的机制还是一样的，把上游与下游重新订阅。

6. 检查异常处理

在java中有一类异常是需要显示进行处理的，那就是检查异常（Checked Exception)，如IOException。在命令式编程中，可以通过throws关键字来声明，从而可以把异常往外抛，而不需要立即处理。然而，遗憾的是，在reactor中，并没有类似的平替，不管任何情况，当遇到检查异常，reactor中都需要用try-catch来处理，这是唯一一个在reactor中没有找到命令式编程中的平替。与命令式编程有throws关键字声明不同，reactor中处理检查异常都必须用try-catch来处理，处理的方式有以下三种：

• 捕获到异常并从中恢复。序列继续正常的进行。
• 捕获异常，将其封装成一个 不检查 的异常，然后将其抛出（中断序列）。
• 如果你需要返回一个 Flux（例如，在 flatMap 中），那么就用一个产生错误的 Flux 来封装异常，如下所示：return Flux.error(checkedException)。（这个序列也会终止。）

这三种方式中，其中最常见也最常用的方式就是第二种，将检查异常转化为非检查异常，如throw new RuntimeException(e)。但是reactor提供了辅助工具类Exceptions，进而可以相对优雅简洁的进行统一处理。

如以下这个例子（来自reactor的文档）：

public String convert(int i) throws IOException {
    if (i > 3) {
        throw new IOException("boom " + i);
    }
    return "OK " + i;
}
Flux<String> converted = Flux
    .range(1, 10)
    .map(i -> {
        try { return convert(i); }
        catch (IOException e) { throw Exceptions.propagate(e); }
    });
converted.subscribe(
    v -> System.out.println("RECEIVED: " + v),
    e -> {
        if (Exceptions.unwrap(e) instanceof IOException) {
            System.out.println("Something bad happened with I/O");
        } else {
            System.out.println("Something bad happened");
        }
    }
);

由于convert声明了检查异常IOException，所以必须要try-catch住，再利用Exceptions.propagate来封装为非检查异常。相比于直接用throw new RuntimeException(e)，利用Exceptions的好处在onError处理阶段可以用Exceptions.unwrap()方法来获取内部真实抛出的异常，体现了利用工具类的好处——简洁明了。

总结

本文先从reactor异常处理的底层机制讲起，讲清楚了一个基本概念：只要出现异常，不管如何处理，旧的流都已经结束，接下来处理的都是新的流。在这基础上，按命令式编程中的try-catch-finally的方式，用reactor的方式进行了一一替代介绍，希望通过对比的方式，能更好的掌握在reactor中如何优雅的处理异常。

到此这篇关于详解Reactor如何优雅Exception异常处理的文章就介绍到这了,更多相关Reactor Exception异常处理内容请搜索码农之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持码农之家！