初识响应式编程的时候，除了从命令式的思维方式转变为函数式的编程方式外，其中有一个很大的不适应的地方就是在面对异常时该怎么处理，尤其是面对检查异常（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.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的源码里可以看到：
在了解了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]
当遇到异常的时候，可以使用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则不会有作用。
打印日志就比较简单了，可以用doOnError方法来实现。doOnError的底层则用的FluxPeek来实现，其作用是覆写了所有的接口，如onNext,onError, cancel等，通过覆写来实现hook。几乎所有doOnXXX的方法都是依赖FluxPeek实现的，例如log、doOnNext、doOnError等等。由于与本次的主题无关，不再赘述，感兴趣的可以自行翻看FluxPeek的实现。
需要注意的是：虽然doOnXXX主要用于打印日志，但如果doOnXXX内部出错，也会导致整个流结束，进入onError阶段。所以，也是有副作用的，仍然在主流程中。
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）。
对于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
);除了以上方式处理异常时，还有一种常见的方式就是重试。比如，我们调用某个接口超时时，通常会重试一次，这个时候可以使用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的源码相对复杂，这里不再叙述（毕竟本文不是源码解读），但除了重试策略有区别外，其重试的机制还是一样的，把上游与下游重新订阅。
在java中有一类异常是需要显示进行处理的，那就是检查异常（Checked Exception)，如IOException。在命令式编程中，可以通过throws关键字来声明，从而可以把异常往外抛，而不需要立即处理。然而，遗憾的是，在reactor中，并没有类似的平替，不管任何情况，当遇到检查异常，reactor中都需要用try-catch来处理，这是唯一一个在reactor中没有找到命令式编程中的平替。与命令式编程有throws关键字声明不同，reactor中处理检查异常都必须用try-catch来处理，处理的方式有以下三种：
这三种方式中，其中最常见也最常用的方式就是第二种，将检查异常转化为非检查异常，如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中如何优雅的处理异常。
