Reactor学习：四、异常处理

声明：

---

数据源：

Flux<String> stringFlux = Flux.just("str1" , "str2" , "str3" , "str4")
        .map(item -> {
            if(item.equals("str3")) throw new RuntimeException("mtk test error");
            LockSupport.parkNanos(Duration.ofSeconds(1).toNanos());
            return item;
        });

一、onErrorReturn

效果：捕获异常并返回一个固定的值
示例：

stringFlux.onErrorReturn(ex -> "mtk test error".equals(ex.getMessage()) , "mtk fallback value").onErrorReturn("hahahaha").subscribe(System.out::println);

结果：

str1
str2
mtk fallback value

由结果可以看到第二个onErrorReturn("hahahaha")并没有被执行，因为错误在第一个时被转换为一个字符串了

---

二、onErrorResume

效果：提供一个异常对象给你，要求你返回一个Publisher的实现类，即你可以根据该异常对象自定义需要返回给下游的东西，实际上大部分方法都是依靠本方法来实现的，比如onErrorReturn、onErrorMap
示例：

stringFlux.onErrorResume(ex -> "mtk test error".equals(ex.getMessage()) , ex -> {
            return Mono.just("error mono");
        }).subscribe(System.out::println);

结果：

str1
str2
error mono

---

三、onErrorMap

效果：将异常转换为一个新的异常
示例：

stringFlux.onErrorMap(ex -> new IllegalStateException("transform the ex")).onErrorResume(ex -> {
    return Mono.just(ex.toString());
}).subscribe(System.out::println);

结果：

str1
str2
java.lang.IllegalStateException: transform the ex

---

四、doOnError

效果：提供一个异常对象，让你自己进行自定义操作，不需要返回。注意doOnError并不会把错误拦截消费掉，错误依然会流入下游，但是由于有doOnError的存在，异常信息并不会被显示报出，即不会在控制台打印出堆栈信息
示例：

stringFlux.doOnError(ex -> System.out.println(ex.getMessage())).onErrorReturn("hahahaha").subscribe(System.out::println);

结果：

str1
str2
mtk test error
hahahaha

---

五、doFinally

效果：提供一个SignalType对象给你，该对象表明从下游来的信号类型，比如request、subscribe、cancel、onNext、onError等等。注意是下游传来的信号！！！
示例1：

stringFlux.doFinally(signalType -> {
    if(signalType.equals(SignalType.CANCEL)){
        System.out.println("cancel");
    }else if(signalType.equals(SignalType.ON_ERROR)){
        System.out.println("error");
    }
}).take(0).subscribe(System.out::println , System.out::println);

结果1：

cancel

示例2：

stringFlux.doFinally(signalType -> {
    if(signalType.equals(SignalType.CANCEL)){
        System.out.println("cancel");
    }else if(signalType.equals(SignalType.ON_ERROR)){
        System.out.println("error");
    }
}).take(3).subscribe(System.out::println , System.out::println);

结果2：

str1
str2
java.lang.RuntimeException: mtk test error
error

---

六、using

效果：类似try-resource-catch，其三个lamada的大致意义为：第一个表示你要生成一个资源，第二个表示你要利用这个资源生成数据源，第三个表示无论最后怎么样，都要对资源执行的操作。
示例：

Flux<String> strFlux = Flux.using(() -> {
    return new String[]{"str1","str2","str3"};
    } , resource -> {
    return Flux.fromArray(resource);
    } , resource -> {
    System.out.println(resource.length);
    }
);
strFlux.handle((item,handler) -> {
    if("str3".equals(item)) throw new IllegalStateException("can not be str3");
    handler.next(item);
}).subscribe(System.out::println);

结果：

str1
str2
3
Exception in thread "main" reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: can not be str3

---

七、retry

效果：如果遇到异常则重新尝试，尝试次数由参数决定（不填表示一直重试），注意重新尝试的根本原理是重新订阅这个流，原来的那个还是被终结掉了。
示例：

stringFlux.elapsed().retry(1).subscribe(System.out::println , System.out::println);

结果：

[1001,str1]
[1001,str2]
[1000,str1]
[1000,str2]
java.lang.RuntimeException: mtk test error

elapsed会把数据封装在一个带有显示“延迟/间隔”的对象中并重新发出，前面的数字正是代表离上一次发出数据用了多久

---

八、Exceptions#propagate和Exceptions#unwrap

效果：前者将流中的异常封装抛出，后者解除封装还原出原异常，常用在受检异常的处理上
示例：

Flux<Long> strFlux = Flux.interval(Duration.ofSeconds(1)).handle((item,handle) -> {
    if(item == 3) {
        try {
            Class.forName("mtk.NoAvailableClass");
        } catch (ClassNotFoundException e) {
            throw Exceptions.propagate(e);
        }
    }else{
        handle.next(item);
    }
});
strFlux.doOnError(System.out::println).subscribe(System.out::println , ex -> {
   if(Exceptions.unwrap(ex) instanceof  ClassNotFoundException){
       System.out.println("找不到类啦!");
   }else{
       System.out.println(ex.getMessage());
   }
});

结果：

0
1
2
java.lang.ClassNotFoundException: mtk.NoAvailableClass
找不到类啦!

---

参考文档：
[1] Reactor api doc
[2] Reactor reference doc