RxJava学习笔记操作符学习（5）

作者: 西红柿的眼泪 | 来源:发表于2018-08-27 09:31 被阅读0次

功能性操作符

作用

辅助被观察者（Observable）在发送事件时实现一些功能性需求,如错误处理、线程调度等等.

类型

类型图.png

*###### 应用场景 & 对应操作符详解
1.subscribe（）
作用
订阅，即连接观察者 & 被观察者，使得被观察者 & 观察者形成订阅关系。

<-- Observable.subscribe(Subscriber) 的内部实现 -->
public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    // 在观察者 subscriber抽象类复写的方法 onSubscribe.call(subscriber)，用于初始化工作
    // 通过该调用，从而回调观察者中的对应方法从而响应被观察者生产的事件
    // 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递，即观察者模式
    // 同时也看出：Observable只是生产事件，真正的发送事件是在它被订阅的时候，即当 subscribe() 方法执行时
}

2.subscribeOn（） & observeOn（）
作用
线程控制，即指定被观察者（Observable） / 观察者（Observer）的工作线程类型。

类型	含义	应用场景
Schedulers.immediate()	当前线程 = 不指定线程	默认
AndroidSchedulers.mainThread()	Android主线程	操作UI
Schedulers.newThread()	常规新线程	耗时等操作
Schedulers.io()	io操作线程	网络请求、读写文件等io密集型操作
Schedulers.computation()	CPU计算操作线程	大量计算操作

注:
若Observable.subscribeOn（）多次指定被观察者生产事件的线程，则只有第一次指定有效，其余的指定线程无效.
.若Observable.observeOn（）多次指定观察者接收 & 响应事件的线程，则每次指定均有效，即每指定一次，就会进行一次线程的切换.
3.delay（）
作用
使得被观察者延迟一段时间再发送事件。

// 1. 指定延迟时间
// 参数1 = 时间；参数2 = 时间单位
delay(long delay,TimeUnit unit)

// 2. 指定延迟时间 & 调度器
// 参数1 = 时间；参数2 = 时间单位；参数3 = 线程调度器
delay(long delay,TimeUnit unit,mScheduler scheduler)

// 3. 指定延迟时间  & 错误延迟
// 错误延迟，即：若存在Error事件，则如常执行，执行后再抛出错误异常
// 参数1 = 时间；参数2 = 时间单位；参数3 = 错误延迟参数
delay(long delay,TimeUnit unit,boolean delayError)

// 4. 指定延迟时间 & 调度器 & 错误延迟
// 参数1 = 时间；参数2 = 时间单位；参数3 = 线程调度器；参数4 = 错误延迟参数
delay(long delay,TimeUnit unit,mScheduler scheduler,boolean delayError): 指定延迟多长时间并添加调度器，错误通知可以设置是否延迟

事例

public void DelayOperators(View view){
        Observable.just(1,2,3)
                .delay(3, TimeUnit.SECONDS)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss","接受到了时间"+integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss","发生了错误事件");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss","事件已经完成");
                    }
                });
    }

4.do（）
作用
在某个事件的生命周期中调用。
do（）操作符有很多个，具体如下：

do的操作符.png

事例

 public void DoOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onError(new Throwable("发生错误了"));
            }
        }).doOnEach(new Consumer<Notification<Integer>>() {
//            当Observable每发送1次数据事件就会调用1次
            @Override
            public void accept(Notification<Integer> integerNotification) throws Exception {
                Log.e("sss","doOnEach"+integerNotification.getValue());
            }
        }).doOnNext(new Consumer<Integer>() {
//            执行Next事件前调用
            @Override
            public void accept(Integer integer) throws Exception {
               Log.e("sss","doOnNext:"+integer);
            }
        }).doAfterNext(new Consumer<Integer>() {
//            执行Next事件后调用
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e("sss","doAfterNext:"+integer);
            }
        }).doOnComplete(new Action() {
//             Observable正常发送事件完毕后调用
            @Override
            public void run() throws Exception {
                Log.e("sss","doOnComplete");
            }
        }).doOnError(new Consumer<Throwable>() {
//             Observable发送错误事件时调用
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.e("sss","doOnError:"+throwable.getMessage());
            }
        }).doOnSubscribe(new Consumer<Disposable>() {
//             观察者订阅时调用
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.e("sss","doOnSubscribe---->");
            }
        }).doAfterTerminate(new Action() {
//            Observable发送事件完毕后调用，无论正常发送完毕 / 异常终止
            @Override
            public void run() throws Exception {
                Log.e("sss","doAfterTerminate");
            }
        }).doFinally(new Action() {
            //  最后执行
            @Override
            public void run() throws Exception {
                Log.e("sss","doFinally:");
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss", "接收到了事件"+ integer  );
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss", "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss", "对Complete事件作出响应");
            }
        });
    }

5.onErrorReturn（）
作用
遇到错误时，发送1个特殊事件 & 正常终止，可捕获在它之前发生的异常。
事例

public void onErrorReturnOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new Throwable("发生了错误"));

            }
        }).onErrorReturn(new Function<Throwable, Integer>() {
            @Override
            public Integer apply(Throwable throwable) throws Exception {
                Log.e("sss", "在onErrorReturn处理了错误: "+throwable.toString() );
                return 888;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss", "接收到了事件"+ integer  );
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss","对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss","对Complete事件作出响应");
            }
        });
    }

6.onErrorResumeNext（）
作用
遇到错误时，发送1个新的Observable，
注：onErrorResumeNext（）拦截的错误 = Throwable；若需拦截Exception请用onExceptionResumeNext（）
若onErrorResumeNext（）拦截的错误 = Exception，则会将错误传递给观察者的onError方法
事例：

  public void onErrorResumeNextOperators(View view){

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new Throwable("发生错误了"));
            }
        }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> apply(Throwable throwable) throws Exception {
                return Observable.just(11,22);
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss","接受到了时间"+integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss","对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss","对Complete事件作出响应");
            }
        });
    }

7.onExceptionResumeNext（）
作用
遇到错误时，发送1个新的Observable
注：
onExceptionResumeNext（）拦截的错误 = Exception；若需拦截Throwable请用onErrorResumeNext（）
若onExceptionResumeNext（）拦截的错误 = Throwable，则会将错误传递给观察者的onError方法
事例

public void onExceptionResumeNextOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new Exception("发生了异常"));
            }
        }).onExceptionResumeNext(new Observable<Integer>() {
            @Override
            protected void subscribeActual(Observer<? super Integer> observer) {
                observer.onNext(11);
                observer.onNext(22);
                observer.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss", "接收到了事件"+ integer  );
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss","对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss","对Complete事件作出响应");
            }
        });
    }

8.retry（）
作用
重试，即当出现错误时，让被观察者（Observable）重新发射数据，即接收到 onError（）时，重新订阅 & 发送事件，Throwable 和 Exception都可拦截。

<-- 1. retry（） -->
// 作用：出现错误时，让被观察者重新发送数据
// 注：若一直错误，则一直重新发送

<-- 2. retry（long time） -->
// 作用：出现错误时，让被观察者重新发送数据（具备重试次数限制
// 参数 = 重试次数
 
<-- 3. retry（Predicate predicate） -->
// 作用：出现错误后，判断是否需要重新发送数据（若需要重新发送& 持续遇到错误，则持续重试）
// 参数 = 判断逻辑

<--  4. retry（new BiPredicate<Integer, Throwable>） -->
// 作用：出现错误后，判断是否需要重新发送数据（若需要重新发送 & 持续遇到错误，则持续重试
// 参数 =  判断逻辑（传入当前重试次数 & 异常错误信息）

<-- 5. retry（long time,Predicate predicate） -->
// 作用：出现错误后，判断是否需要重新发送数据（具备重试次数限制
// 参数 = 设置重试次数 & 判断逻辑

事例

 public void retryOperators(View view){
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

            }
        }).retry().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss", "接收到了事件"+ integer  );
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss","对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.e("sss","对Complete事件作出响应");
            }
        });

    }

9.retryUntil（）
作用
出现错误后，判断是否需要重新发送数据。
10.retryWhen（）
作用
遇到错误时，将发生的错误传递给一个新的被观察者（Observable），并决定是否需要重新订阅原始被观察者（Observable）& 发送事件。

  public void retryWhenOperators(View view){

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onError(new Exception("发生错误了"));
                e.onNext(3);
            }
        })
                // 遇到error事件才会回调
                .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {

                    @Override
                    public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
                        // 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
                        // 返回Observable<?> = 新的被观察者 Observable（任意类型）
                        // 此处有两种情况：
                        // 1. 若 新的被观察者 Observable发送的事件 = Error事件，那么 原始Observable则不重新发送事件：
                        // 2. 若 新的被观察者 Observable发送的事件 = Next事件 ，那么原始的Observable则重新发送事件：
                        return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                            @Override
                            public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {

                                // 1. 若返回的Observable发送的事件 = Error事件，则原始的Observable不重新发送事件
                                // 该异常错误信息可在观察者中的onError（）中获得
                                return Observable.error(new Throwable("retryWhen终止啦"));

                                // 2. 若返回的Observable发送的事件 = Next事件，则原始的Observable重新发送事件（若持续遇到错误，则持续重试）
                                // return Observable.just(1);
                            }
                        });

                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }
                    @Override
                    public void onNext(Integer value) {
                        Log.e("sss", "接收到了事件"+ value  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应" + e.toString());
                        // 获取异常错误信息
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });

    }

11.repeat（）
作用
无条件地、重复发送被观察者事件
事例

    public void repeatOperators(View view){
        Observable.just(1,2,3,4)
                .repeat(3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("sss", "接收到了事件" + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("sss", "对Error事件作出响应" + e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.e("sss", "对Complete事件作出响应");
                    }
                });
    }

12.repeatWhen（）
作用
有条件地、重复发送被观察者事件
实例

 public void repeatWhenOperators(View view){
        Observable.just(1,2,3)
                .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
                        return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {

                            @Override
                            public ObservableSource<?> apply(Object o) throws Exception {
                                return Observable.just(2);
                            }
                        });
                    }
                }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("sss", "开始采用subscribe连接");
            }

            @Override
            public void onNext(Integer integer) {
                Log.e("sss", "接收到了事件" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.e("sss", "对Error事件作出响应：" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.e("sss", "对Complete事件作出响应");
            }
        });
    }

实际开发中的应用

1.有条件网络请求轮询
repeatWhen（）的使用

    int i = 0;
    public void webPolling(View view) {

        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build();
             GetRequest_Interface request_interface = retrofit.create(GetRequest_Interface.class);

              Observable<Translation> observable = request_interface.getCall();

        observable.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {


                return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {

                    @Override
                    public ObservableSource<?> apply(Object o) throws Exception {
                        if (i > 3) {
                            return Observable.error(new Throwable("轮询结果"));
                        }
                        return Observable.just(1).delay(2000, TimeUnit.MILLISECONDS);
                    }
                });
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Translation>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Translation translation) {
                        i++;
                        Log.e("sss", translation.getContent().getOut());

                    }

                    @Override
                    public void onError(Throwable e) {

                        Log.e("sss", e.toString());
                    }

                    @Override
                    public void onComplete() {

                    }
                });
        
    }

2.网络请求出错重连

 // 设置变量
    // 可重试次数
    private int maxConnectCount = 10;
    // 当前已重试次数
    private int currentRetryCount = 0;
    // 重试等待时间
    private int waitRetryTime = 0;

    public void requestRetry(View view){
        Retrofit retrofit=new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build();

        final GetRequest_Interface request=retrofit.create(GetRequest_Interface.class);

        Observable<Translation> observable=request.getCall();

        observable.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
                // 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {

                        // 输出异常信息
                        Log.e("sss",  "发生异常 = "+ throwable.toString());

                        /**
                         * 需求1：根据异常类型选择是否重试
                         * 即，当发生的异常 = 网络异常 = IO异常 才选择重试
                         */
                        if (throwable instanceof IOException){

                            Log.e("sss", "属于IO异常，需重试" );

                            /**
                             * 需求2：限制重试次数
                             * 即，当已重试次数 < 设置的重试次数，才选择重试
                             */
                            if (currentRetryCount < maxConnectCount){

                                // 记录重试次数
                                currentRetryCount++;
                                Log.e("sss",  "重试次数 = " + currentRetryCount);

                                /**
                                 * 需求2：实现重试
                                 * 通过返回的Observable发送的事件 = Next事件，从而使得retryWhen（）重订阅，最终实现重试功能
                                 *
                                 * 需求3：延迟1段时间再重试
                                 * 采用delay操作符 = 延迟一段时间发送，以实现重试间隔设置
                                 *
                                 * 需求4：遇到的异常越多，时间越长
                                 * 在delay操作符的等待时间内设置 = 每重试1次，增多延迟重试时间1s
                                 */
                                // 设置等待时间
                                waitRetryTime = 1000 + currentRetryCount* 1000;
                                Log.e("sss",  "等待时间 =" + waitRetryTime);
                                return Observable.just(1).delay(waitRetryTime, TimeUnit.MILLISECONDS);


                            }else{
                                // 若重试次数已 > 设置重试次数，则不重试
                                // 通过发送error来停止重试（可在观察者的onError（）中获取信息）
                                return Observable.error(new Throwable("重试次数已超过设置次数 = " +currentRetryCount  + "，即 不再重试"));

                            }
                        }

                        // 若发生的异常不属于I/O异常，则不重试
                        // 通过返回的Observable发送的事件 = Error事件 实现（可在观察者的onError（）中获取信息）
                        else{
                            return Observable.error(new Throwable("发生了非网络异常（非I/O异常）"));
                        }
                    }
                });
            }
        }).subscribeOn(Schedulers.io())               // 切换到IO线程进行网络请求
                .observeOn(AndroidSchedulers.mainThread())  // 切换回到主线程 处理请求结果
                .subscribe(new Observer<Translation>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(Translation result) {
                        // 接收服务器返回的数据
                        Log.e("sss",  "发送成功");
                        Log.e("sss",result.getContent().getOut());
//                        result.show();
                    }

                    @Override
                    public void onError(Throwable e) {
                        // 获取停止重试的信息
                        Log.e("sss",  e.toString());
                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

RxJava学习笔记操作符学习（5）

功能性操作符

作用

类型

实际开发中的应用

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读