RxJava——基础学习(八)，错误操作符

作者: 英勇青铜5 | 来源:发表于2016-12-12 16:56 被阅读49次

学习资料

ReactiveX文档中文翻译

错误操作符不多，文档上暂时是就两种，但感觉蛮实用的，但两种操作符有好多种延伸

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

学习资料：

小鄧子大神的【译】对RxJava中.repeatWhen()和.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有效期为半个小时，后来改成了一个星期，还没具体实践过

本人很菜，有错误请指出

共勉：）

RxJava——基础学习(八)，错误操作符

1. Retry重试

1.1 retry()

1.2 retry(long)

1.3 retry(Func2)

1.4 retryWhen

2. Catch

2.1 onErrorReturn

2.2 onErrorResumeNext

2.3 onExceptionResumeNext

3. 最后

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Android-RxJava

Android-Rxjava&retrofit&dagger