RxJava——基础学习(八)，错误操作符
2016-12-12 本文已影响49人
英勇青铜5
学习资料
错误操作符不多，文档上暂时是就两种，但感觉蛮实用的，但两种操作符有好多种延伸
1. Retry重试
当原始Observable_Ob遇到错误时，，接收到onError信号时，会重新订阅Ob，直到能正常进行操作终止,操作符默认在trampoline调度器上执行
而repeat是接到onCompleted成功信号后，重复订阅
retry是失败后再次订阅，repeat是成功后重复再来一遍
1.1 retry()
只要收到onError()的通知，就会重新订阅并发射原始Observable，无论多少次。
简单使用：
/**
* 当为4时，制造一个异常，重新开始，就重复
*/
private static void retry() {
Observable
.just(1, 2, 3, 4, 5)
//将数据进行变换 小于4就正常发送
.map((i) -> (i == 4) ? (i / 0) : (i))
//出现异常，重复
.retry()
//对发送数据项进行限制，就输出10个，否则会无限输出
.limit(10)
//将3个分一组，缓存进List，方便观察结果
.buffer(3)
.subscribe(new Subscriber<List<Integer>>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println(e.getMessage());
}
@Override
public void onNext(List<Integer> integers) {
System.out.println(integers);
}
});
}
运行结果：
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1]
onCompleted
结束时，会得到onCompleted信号，并不会得到异常信息
1.2 retry(long)
- retry(count)
当收到onError()的通知时，最多重复进行count次订阅
简单使用：
/**
* 出现异常，就重复订阅5次
*/
private static void retryCount() {
Observable
.just(1, 2, 3, 4, 5)
//将数据进行变换
.map((i) -> (i == 4) ? (i / 0) : (i))
//重复5次
.retry(5)
//3个数据一组
.buffer(3)
.subscribe(new Subscriber<List<Integer>>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println(e.getMessage());
}
@Override
public void onNext(List<Integer> integers) {
System.out.println(integers);
}
});
}
运行结果：
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
/ by zero
当重复订阅次数结束时，如果异常仍然为解决，便会抛出异常，可以拿到异常的信息，不会得到onCompleted信号
1.3 retry(Func2)
传入一个Func2，通过函数来控制更加定制化的重复订阅
/**
* 传入一个函数，可以控制重复订阅次数以及是否发送onError信号通知
*/
private static void retryFunc() {
Observable
.just(1, 2, 3, 4, 5)
//将数据进行变换
.map((i) -> (i == 4) ? (i / 0) : (i))
.retry(new Func2<Integer, Throwable, Boolean>() {
//integer 已重复次数
//throwable 异常
@Override
public Boolean call(Integer integer, Throwable throwable) {
//System.out.println(integer + throwable.getMessage());
//返回结果 true 重复订阅； false ，不再订阅，并将onError通知发送给订阅者；null，不再订阅，不接收onError通知
//重复次数小于5 就再次订阅，超过5次，就结束，不再订阅
return integer<= 5 ? true :null;
}
})
.buffer(3)
.subscribe(new Subscriber<List<Integer>>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println(e.getMessage());
}
@Override
public void onNext(List<Integer> integers) {
System.out.println(integers);
}
});
}
运行结果：
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
[1, 2, 3]
1.4 retryWhen
学习资料：
retryWhen和retry功能类似，都是在接到 onError信号通知的时候进行重新订阅，默认在trampoline调度器上执行
retryWhen会将作为onError信号的Throwable传递给一个函数，函数会返回一个新的Observable，新的Observable发送出一个数据项，retryWhen就会重新订阅原始的Observable；如果新的Observable发出了onError信号，则不再重新订阅
retryWhen
简单使用：
/**
* 出现异常时，重复订阅5次，并延迟 2 ^ integer 秒订阅
*/
private static void retryWhen() {
Observable
.just(1, 2, 3, 4, 5)
.map((i) -> (i == 4) ? (i / 0) : (i))
.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override
public Observable<?> call(Observable<? extends Throwable> observable) {
return observable
.zipWith(Observable.range(1, 5), new Func2<Throwable, Integer, Integer>() {
@Override
public Integer call(Throwable throwable, Integer i) {
System.out.println(i + " -- " + throwable.getMessage());
return i;
}
})
.flatMap(new Func1<Integer, Observable<?>>() {
@Override
public Observable<?> call(Integer integer) {
//System.out.println(Math.pow(2, integer));
return Observable
.timer((long) Math.pow(2, integer), TimeUnit.SECONDS, Schedulers.immediate());
}
});
}
})
.buffer(3)
.subscribe(new Subscriber<List<Integer>>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println(e.getMessage());
}
@Override
public void onNext(List<Integer> integers) {
System.out.println(integers);
}
});
}
运行结果：
1 -- / by zero
[1, 2, 3]
2 -- / by zero
[1, 2, 3]
3 -- / by zero
[1, 2, 3]
4 -- / by zero
[1, 2, 3]
5 -- / by zero
onCompleted
在小鄧子大神博客最后提到一个二进制指数退避算法，Math.pow(2, integer)，延迟2^integer秒，重复订阅次数越多，延迟就越久
算法规则：
- 对每个数据帧，当第一次发生冲突时，设置一个参数L=2
- 退避间隔取1到L个时间片中的一个随机数，一个时间片等于两个节点之间最大传播时延的两倍
- 当数据帧再次发生冲突，则将参量L加倍
- 设置一个最大重传次数，超过该次数，则不再重传，并报告出错
二进制指数退避算法是按后进先出的次序控制的，即未发生冲突或很少发生冲突的数据帧，具有优先发送的概率；而发生过多次冲突的数据帧，发送成功的概率就更小
每个字都认识，组在一起，不懂，先记下来，知道有这么个玩意
2. Catch
操作符拦截原始Observable的onError通知，将onError通知替换为其它的数据项或数据序列，让产生的Observable能够正常终止或者根本不终止
2.1 onErrorReturn
在接到onError错误通知时，发送一个特殊的数据项后立即正常结束
简单使用：
/**
* 出现异常时，返回10
*/
private static void catchReturnf() {
Observable
.just(1, 2, 3, 4, 5)
.map((i) -> (i == 4) ? (i / 0) : (i))
.onErrorReturn(new Func1<Throwable, Integer>() {
@Override
public Integer call(Throwable throwable) {
return 10;
}
})
.buffer(3)
.subscribe(System.out::println);
}
运行结果：
[1, 2, 3]
[10]
2.2 onErrorResumeNext
原始Observable遇到错误发送onError关闭信号时，通过一个函数创建一个新的Observable
简单使用：
/**
* 遇到异常时，发送一个新的Observable的数据项
*/
private static void catchErrorNext() {
Observable
.just(1, 2, 3, 4, 5)
.map((i) -> (i == 4) ? (i / 0) : (i))
.onErrorResumeNext(new Func1<Throwable, Observable<? extends Integer>>() {
@Override
public Observable<? extends Integer> call(Throwable throwable) {
return Observable.just(6,7,8,9);
}
})
.buffer(3)
.subscribe(System.out::println);
}
运行结果：
[1, 2, 3]
[6, 7, 8]
[9]
2.3 onExceptionResumeNext
在接到onError错误信号Throwable_T时进行判断：
- 若
T是一个Exception异常 不是一个错误就会启用onExceptionResumeNext新生成的备用Obsvable - 若
T是一个Error错误，则将这个错误发送到订阅者的onError()方法，不会使用新生成的备用Observable
简单使用：
/**
* 异常时，启用备用Observable
*/
private static void catchExceptionNext() {
Observable
.just(1, 2, 3, 4, 5)
.map((i) -> (i == 4) ? (i / 0) : (i))
.onExceptionResumeNext(Observable.just(6,7,8))
.buffer(3)
.subscribe(System.out::println);
}
运行结果：
[1, 2, 3]
[6, 7, 8]
3. 最后
重点是retryWhen的理解，在处理一些过期token时，会用到。我们的项目中原本token有效期为半个小时，后来改成了一个星期，还没具体实践过
本人很菜，有错误请指出
共勉 ：）
