Android :RxJava学习笔记之 错误处理

Posted JMW1407

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Android :RxJava学习笔记之 错误处理相关的知识,希望对你有一定的参考价值。

错误处理

概述

熟悉RxJava的知道,onError跟onComplete是互斥的,出现其中一个,观察者与被观察者的关系就被中断(以下简称:管道中断),观察者就永远不会收到来自被观察者发出的事件。

然后有些情况下,出现了错误,我们希望可以进行一些补救措施,例如:

  • 由于网络原因或者其他原因,Http请求失败了,这个时候我们希望进行重试,又或者去读取本地的缓存数据
  • 在使用RxJava的组合操作符进行Http并发请求时,我们希望接口之间互不影响,即A接口出现错误不会影响B接口的正常流程,反之一样

现实开发中,可能有更多的场景需要对错误进行补救,所以RxJava为我们提供了两大类进行错误处理,分别是Catch和Retry,前者在出现错误时补救,后者在出现错误时重试,接下来,分别对它们进行讲解

注:Catch和Retry只能捕获上游事件的异常

1. Catch

onErrorReturnItem(final T item)  //内部调用第二个方法
onErrorReturn(Function function)  //遇到错误,用默认数据项替代
onErrorResumeNext(final ObservableSource<? extends T> next) //内部调用第四个方法
onErrorResumeNext(Function resumeFunction ) //遇到错误,开始发射新的Observable的数据序列
onExceptionResumeNext(final ObservableSource<? extends T> next) //内部原理与第四个相同,仅有一个参数不同

虽然有5个操作符,但是实际上就只有3个,再准确点说就只有2个,为什么这么说呢?

1、因为第1个操作符内部调用的就是第2个,而第3个操作符内部调用是第4个操作符,所以说只有3个。

2、那为什么准确点说只有2个呢,因为第5个操作符,内部原理同第四个,仅仅有一个参数传的不一样,接下来我们分别讲解。

1.1 onErrorReturnItem

onErrorReturnItem(T item): 让Observable遇到错误时发射一个指定的项(item)并且正常终止。

Disposable disposable = Observable
        .fromCallable(new Callable<Integer>() 
            @Override
            public Integer call() throws Exception 
                return null; //返回null,即出现错误
            
        )
        .onErrorReturnItem(100) //出现错误时,用一个默认的数据项将其替代
        .subscribe(new Consumer<Integer>() 
            @Override
            public void accept(Integer integer) throws Exception 
                // 打印 100   随后会立即回调onComplete
            
        );

我们再来看看onErrorReturnItem内部实现

public final Observable<T> onErrorReturnItem(final T item) 
    ObjectHelper.requireNonNull(item, "item is null");
    //将item封装成Function对象,并调用onErrorReturn方法
    return onErrorReturn(Functions.justFunction(item));

1.2 onErrorReturn

onErrorReturn(Function<Throwable, T> valueSupplier):让Observable遇到错误时通过一个函数Function来进行判断返回指定的类型数据,并且正常终止。

Disposable disposable = Observable
        .fromCallable(new Callable<Integer>() 
            @Override
            public Integer call() throws Exception 
                return null; //返回null,即出现错误
            
        )
        .onErrorReturn(new Function<Throwable, Integer>() 
            @Override
            public Integer apply(Throwable throwable) throws Exception 
                //出现错误时,用一个默认的数据项将其替代,这里根据不同的错误返回不同的数据项
                return 100; 
            
        ) 
        .subscribe(new Consumer<Integer>() 
            @Override
            public void accept(Integer integer) throws Exception 
                // 打印 100  随后会立即回调onComplete
            
        );

看到上面代码可以明白,onErrorReturn的作用就是在出现错误的时候,用一个默认的数据项将错误替代,并立刻回调onComplete。

    // 创建一个可以发射异常的Observable
    Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() 
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception 
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(1 / 0);  // 产生一个异常
            emitter.onNext(3);
            emitter.onNext(4);
        
    );

    /** 1. onErrorReturnItem(T item)
     * 让Observable遇到错误时发射一个指定的项(item)并且正常终止。
     */
    observable.onErrorReturnItem(888)   // 源Observable发生异常时发射指定的888数据
            .subscribe(new Observer<Integer>() 
                @Override
                public void onSubscribe(Disposable d) 
                    System.out.println("--> onSubscribe(1)");
                

                @Override
                public void onNext(Integer integer) 
                    System.out.println("--> onNext(1): " + integer);
                

                @Override
                public void onError(Throwable e) 
                    System.out.println("--> onError(1): " + e);
                

                @Override
                public void onComplete() 
                    System.out.println("--> onCompleted(1)");
                
            );

    System.out.println("-----------------------------------------------");
    /**
     * 2. onErrorReturn(Function<Throwable, T> valueSupplier)
     * 让Observable遇到错误时通过一个函数Function来接受Error参数并进行判断返回指定的类型数据,并且正常终止。
     */
    observable.onErrorReturn(new Function<Throwable, Integer>() 
        @Override
        public Integer apply(Throwable throwable) throws Exception 
            System.out.println("--> apply(1): e = " + throwable);
            return 888; // 源Observable发生异常时发射指定的888数据
        
    ).subscribe(new Observer<Integer>() 
        @Override
        public void onSubscribe(Disposable d) 
            System.out.println("--> onSubscribe(2)");
        

        @Override
        public void onNext(Integer integer) 
            System.out.println("--> onNext(2): " + integer);
        

        @Override
        public void onError(Throwable e) 
            System.out.println("--> onError(2): " + e);
        

        @Override
        public void onComplete() 
            System.out.println("--> onCompleted(2)");
        
    );

输出

--> onSubscribe(1)
--> onNext(1): 1
--> onNext(1): 2
--> onNext(1): 888
--> onCompleted(1)
-----------------------------------------------
--> onSubscribe(2)
--> onNext(2): 1
--> onNext(2): 2
--> apply(1): e = java.lang.ArithmeticException: / by zero
--> onNext(2): 888
--> onCompleted(2)

1.3 onErrorResumeNext

  • onErrorResumeNext(ObservableSource next):让Observable在遇到错误时开始发射第二个指定的Observable的数据序列。
  • onErrorResumeNext(Function<Throwable, ObservableSource>resumeFunction):让Observable在遇到错误时通过一个函数Function来接受Error参数并进行判断返回指定的第二个Observable的数据序列。

onErrorResumeNext 方法返回一个镜像原有Observable行为的新Observable,后者会忽略前者的 onError 调用,不会将错误传递给观察者,作为替代,它会开始另一个指定的备用Observable。


onErrorResumeNext(final ObservableSource next)

public final Observable<T> onErrorResumeNext(final ObservableSource<? extends T> next) 
    ObjectHelper.requireNonNull(next, "next is null");
    //可以看到这里将Observable对象封装成Function对象,并调用onErrorResumeNext方法
    return onErrorResumeNext(Functions.justFunction(next));

onErrorResumeNext(final ObservableSource next)内部调用了onErrorResumeNext(Function resumeFunction )

onErrorResumeNext(Function resumeFunction )

onErrorResumeNext的作用就是在遇到错误时开始发射第二个Observable的数据序列

Disposable disposable = Observable
        .fromCallable(new Callable<Integer>() 
            @Override
            public Integer call() throws Exception 
                return null; //返回null,即出现错误
            
        )
        .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Integer>>() 
            @Override
            public ObservableSource<? extends Integer> apply(Throwable throwable) throws Exception 
                //出现错误时开始发射新的Observable的数据序列
                return Observable.just(1, 2, 3);
            
        )
        .subscribe(new Consumer<Integer>() 
            @Override
            public void accept(Integer integer) throws Exception 
                // 打印 1、2、3  随后会立即回调onComplete
            
        );

1.4 onExceptionResumeNext

onExceptionResumeNextonErrorResumeNext作用相同,都是在遇到错误时开始发射第二个Observable的数据序列,不同的是,如果onError收到的Throwable不是一个Exception,它会将错误传递给观察者的onError方法,不会使用备用的Observable,即它只能捕获Exception异常,这一点,我们可以在ObservableOnErrorNext$OnErrorNextObserver类中源码看到

//使用onExceptionResumeNext操作符时,allowFatal为true
//使用onErrorResumeNext操作都是,allowFatal为false
if (allowFatal && !(t instanceof Exception)) 
    //非Exception异常,直接交给观察者的onError方法
    actual.onError(t);                        
    return;                                   
                                             


解析: onExceptionResumeNext 只会对Exception类型的异常进行处理,如果onError收到的Throwable不是一个Exception,它会将错误传递给观察者的onError方法,不会使用备用的Observable

    // 创建一个可以发射异常的Observable
    Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() 
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception 
            emitter.onNext(1);
            emitter.onNext(2);
        //  emitter.onError(new Throwable("This is Throwable!"));  // Throwable类型异常,直接通知观察者
        //  emitter.onError(new Error("This is Error!"));          // Error类型异常,直接通知观察者
            emitter.onError(new Exception("This is Exception!"));  // Exception类型异常,进行处理,发送备用的Observable数据
        //    emitter.onNext(1 / 0);  // 会产生一个ArithmeticException异常,异常会被处理,发送备用的Observable数据
            emitter.onNext(3);
            emitter.onNext(4);
        
    );
    /**
     * 5. onExceptionResumeNext(ObservableSource next)
     *  如果onError收到的Throwable不是一个Exception,它会将错误传递给观察者的onError方法,不会使用备用的Observable
     *  只对Exception类型的异常通知进行备用Observable处理
     */
    observable1.onExceptionResumeNext(Observable.just(888))
            .subscribe(new Observer<Integer>() 
                @Override
                public void onSubscribe(Disposable d) 
                    System.out.println("--> onSubscribe(5)");
                

                @Override
                public void onNext(Integer integer) 
                    System.out.println("--> onNext(5): " + integer);
                

                @Override
                public void onError(Throwable e) 
                    System.out.println("--> onError(5): " + e);
                

                @Override
                public void onComplete() 
                    System.out.println("--> onCompleted(5)");
                
            );

输出

--> onSubscribe(5)
--> onNext(5): 1
--> onNext(5): 2
--> onNext(5): 888
--> onCompleted(5)

2. Retry

Retry顾名思义就是在出现错误的时候进行重试,共有7个操作符,如下

retry()                                //无条件,重试无数次
retry(long times)                      //无条件,重试times次
retry(Predicate predicate)             //根据条件,重试无数次
retryUntil(final BooleanSupplier stop) //根据条件,重试无数次
retry(long times, Predicate predicate) //根据条件,重试times次
retry(BiPredicate predicate)           //功能与上一个一样,实现不同
retryWhen(final Function handler)      //可以实现延迟重试n次

前4个操作符内部都调用第五个retry(long times, Predicate predicate)(需要注意的是retryUntil操作符,只有接口里的方法返回false时,才会重试),所以我们直接从第五个开始

2.1 retry()

retry():无论收到多少次 onError 通知,无参数版本的 retry 都会继续订阅并发射原始Observable。

注意: 因为如果遇到异常,将会无条件的重新订阅原始的Observable,知道没有异常的发射全部的数据序列为止。所以如果你的异常发生后重新订阅也不会恢复正常的话,会一直订阅下去,有内存泄露的风险。

2.2 retry(long times)

retry(long times):接受单个 count 参数的 retry 会最多重新订阅指定的次数,如果次数超了,它不会尝试再次订阅,它会把最新的一个 onError 通知传递给它的观察者。

2.3 retry(long times, Predicate predicate)

retry(long times, Predicate<Throwable> predicate):遇到异常后最多重新订阅 times 次,每次重新订阅经过函数predicate 最终判断是否继续重新订阅,如果 times 到达上限或者 predicate 返回 false 中任意一个最先满足条件,都会终止重新订阅,retry 会将最新的一个 onError 通知传递给它的观察者。

Disposable disposable = Observable
        .fromCallable(new Callable<Integer>() 
            @Override
            public Integer call() throws Exception 
                //这里会执行n+1次,其中n为重试次数(如果重试条件为true的话)
                return null; //返回null,即出现错误
            
        )
        .retry(3, new Predicate<Throwable>() //重试3次
            @Override
            public boolean test(Throwable throwable) throws Exception 
                //true 代表需要重试,可根据throwable对象返回是否需要重试
                return true;
            
        )
        .subscribe(new Consumer<Integer>() 
            @Override
            public void accept(Integer integer) throws Exception 
                //重试成功,走这里
            
        , new Consumer<Throwable>() 
            @Override
            public void accept(Throwable throwable) throws Exception 
                //重试n次后,依然出现错误,直接会走到这里
            
        );

2.4 retry(Predicate predicate)

retry(Predicate predicate):接受一个谓词函数作为参数,这个函数的两个参数是:重试次数和导致发射 onError 通知的 Throwable 。这个函数返回一个布尔值,如果返回 true , retry 应该再次订阅和镜像原始的Observable,如果返回 false , retry 会将最新的一个 onError 通知传递给它的观察者

Disposable disposable = Observable
        .fromCallable(new Callable<Integer>() 
            @Override
            public Integer call() throws Exception 
                //这里会执行n+1次,其中n为重试次数(如果重试条件为true的话)
                return null; //返回null,即出现错误
            
        )
        .retry(new BiPredicate<Integer, Throwable>() 
            @Override
            public boolean test(Integer times, Throwable throwable) throws Exception 
                //times 为尝试次数,即第几次尝试
                return times <= 3; //只允许尝试3次
            
        )
        .subscribe(new Consumer<Integer>() 
            @Override
            public void accept(Integer integer) throws Exception 
                //重试成功,走这里
            
        , new Consumer<Throwable>() 
            @Override
            public void accept(Throwable throwable) throws Exception 
                //重试n次后,依然出现错误,直接会走到这里
            
        );

retry(BiPredicate predicate)跟retry(long times, Predicate predicate)功能上是一样的,只是实现不一样而已。注释很详情,也不再讲解。

2.5 retry(BiPredicate predicate)

retry(BiPredicate<Integer, Throwable> predicate):遇到异常时,通过函数 predicate 判断是否重新订阅源Observable,并且通过参数 Integer 传递给 predicate 重新订阅的次数,retry 会将最新的一个 onError 通知传递给它的观察者。

2.6 retryUntil(BooleanSupplier stop)

retryUntil(BooleanSupplier stop):重试重新订阅,直到给定的停止函数 stop 返回 true,retry 会将最新的一个 onError 通知传递给它的观察者。

2.7 retryWhen(Function handler)

retryWhen(Function<Observable, ObservableSource> handler):retryWhen 和 retry 类似,区别是, retryWhen 将 onError 中的 Throwable 传递给一个函数,这个函数产生另一个 Observable, retryWhen 观察它的结果再决定是不是要重新订阅原始的Observable。如果这个Observable发射了一项数据,它就重新订阅,如果这个Observable发射的是 onError 通知,它就将这个通知传递给观察者然后终止。


retryWhen将onError中的Throwable传递给一个函数,这个函数产生另一个Observable,retryWhen观察它的结果再决定是不是要重新订阅原始的Observable。如果这个Observable发射了一项数据,它就重新订阅,如果这个Observable发射的是onError通知,它就将这个通知传递给观察者然后终止。

这段话的大致意思就是,如果RxJava内部传过来的Observable(retryWhen方法传入的接口,通过接口方法传过来的)发射了一项数据,即发射onNext事件,就会重新订阅原始的Observable,如果发射的是onError事件,它就将这个事件传递给观察者然后终止。

那么,retryWhen有什么作用呢,它的主要作用出现错误时,重新订阅,即重试,它跟之前的retry操作符最大的区别就是,它可以延迟重试,例如,我们有这样一个需求,需要在遇到错误是,隔3秒重试一次,最多重试3次。

Disposable disposable = Observable
        .fromCallable(new Callable<Integer>() 
            @Override
            public Integer call() throws Exception 
                Log.d("LJX", "call");
                //这里会执行n+1次,其中n为重试次数
                return null; //返回null,即出现错误
            
        )
        .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() 
            @Override
            public ObservableSource<?> apply(Observable<Throwable> attempts) throws Exception 
                //注:这里需要根据RxJava传递过来的Observable对象发射事件,不能直接返回一个新的Observable,否则无效
                return attempts.flatMap(new Function<Throwable, ObservableSource<以上是关于Android :RxJava学习笔记之 错误处理的主要内容,如果未能解决你的问题,请参考以下文章

Android :RxJava学习笔记之SingleCompletable以及Maybe

Android :RxJava学习笔记之合并操作符

Android :RxJava学习笔记之过滤操作符

Android :RxJava学习笔记之条件/布尔操作符

Android :RxJava学习笔记之Subject

Android :RxJava学习笔记之创建操作符