声明：
数据源：
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
网友评论