RxJava操作符-->创建

作者: 谢尔顿 | 来源:发表于2018-06-14 13:15 被阅读23次

引言

该篇文章主要是关于RxJava的创建操作符使用的代码讲解。关于创建被观察者操作符共有以下几种：

create()
just()
fromArray()
fromIterable()
never()
empty()
error()
defer()
timer()
interval()
intervalRange()
range()
rangeLong()

1. create()

该操作符已经在文章中使用过了，关于其的具体使用，可以自己点击查看。RxJava实例-->基本使用

2.just()

快速创建被观察者对象（Observable），最多只能10个参数。

        // 在创建后就会发送这些对象，相当于执行了onNext(1)、onNext(2)、onNext(3)、onNext(4)
        Observable.just(1, 2, 3, 4,)
                .subscribe(new Observer<Integer>() {
                    //观察者接收事件前，默认最先调用复写onSubscribe()
                    @Override
                    public void onSubscribe(Disposable d) {
                        Logger.e("开始采用subscribe连接");
                    }

                    //当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onNext(Integer value) {
                        Logger.e("对Next事件作出响应" + value);
                    }

                    //当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onError(Throwable e) {
                        Logger.e("对Error事件作出响应");
                    }

                    //当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onComplete() {
                        Logger.e("对Complete事件作出响应");
                    }
                });

3. fromArray()

快速创建被观察者对象（Observable），能发送10个以上事件（数组形式）

        //设置需要传入的数组
        Integer[] items = {0,1,2,3,4,5,6};
        //创建被观察者对象时传入数组
        Observable.fromArray(items)
                .subscribe(new Observer<Integer>() {
                    //观察者接收事件前，默认最先调用复写onSubscribe()
                    @Override
                    public void onSubscribe(Disposable d) {
                        Logger.e("数组遍历");
                    }

                    //当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onNext(Integer value) {
                        Logger.e("数组中的元素 = " + value);
                    }

                    //当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onError(Throwable e) {
                        Logger.e("对Error事件作出响应");
                    }

                    //当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onComplete() {
                        Logger.e("遍历结束");
                    }
                });

4. fromIterable()

快速创建被观察者对象（Observable）发送10个以上事件（集合形式）

ArrayList<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        list.add(4);
        Observable.fromIterable(list)
                .subscribe(new Observer<Integer>() {
                    //观察者接收事件前，默认最先调用复写onSubscribe()
                    @Override
                    public void onSubscribe(Disposable d) {
                        Logger.e("开始采用subscribe连接");
                    }

                    //当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onNext(Integer value) {
                        Logger.e("对Next事件作出响应" + value);
                    }

                    //当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onError(Throwable e) {
                        Logger.e("对Error事件作出响应");
                    }

                    //当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
                    @Override
                    public void onComplete() {
                        Logger.e("对Complete事件作出响应");
                    }
                });

5. never()

被观察者不发送任何事件。

6. empty()

被观察者仅发送complete事件。观察者会回调onComplete方法。

       Observable.empty().subscribe(new Observer<Object>() {
        //观察者接收事件前，默认最先调用复写onSubscribe()
        @Override
        public void onSubscribe(Disposable d) {
            Logger.e("开始采用subscribe连接");
        }

        //当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
        @Override
        public void onNext(Object value) {
            Logger.e("对Next事件作出响应");
        }

        //当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
        @Override
        public void onError(Throwable e) {
            Logger.e("对Error事件作出响应");
        }

        //当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
        @Override
        public void onComplete() {
            Logger.e("对Complete事件作出响应");
        }
        });

log信息：

03-04 21:26:15.608 2844-2844/com.gjj.frame E/gjj: 开始采用subscribe连接
03-04 21:26:15.608 2844-2844/com.gjj.frame E/gjj: 对Complete事件作出响应

7. error()

被观察者仅发送error事件。

        Observable.error(new RuntimeException()).subscribe(new Observer<Object>() {
        //观察者接收事件前，默认最先调用复写onSubscribe()
        @Override
        public void onSubscribe(Disposable d) {
            Logger.e("开始采用subscribe连接");
        }

        //当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
        @Override
        public void onNext(Object value) {
            Logger.e("对Next事件作出响应");
        }

        //当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
        @Override
        public void onError(Throwable e) {
            Logger.e("对Error事件作出响应");
        }

        //当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
        @Override
        public void onComplete() {
            Logger.e("对Complete事件作出响应");
        }
        });

log信息：

03-04 21:30:13.520 4530-4530/com.gjj.frame E/gjj: 开始采用subscribe连接
03-04 21:30:13.520 4530-4530/com.gjj.frame E/gjj: 对Error事件作出响应

8. defer()

直到有观察者（Observer）订阅时，才动态创建被观察对象（Observable）以及发送事件

    //第一次对i的赋值
    private int i= 10;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_defer);

        //通过defer定义被观察者对象，此时观察者对象孩没创建
        Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> call() throws Exception {
                return Observable.just(i);
            }
        });

        //第二次对i的赋值
        i =15;

        //观察者开始订阅，此时才会调用defer()创建被观察者对象（Observable）
        observable.subscribe(new Observer<Integer>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG,"开始采用subscribe连接");
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG,"接收到的整数是"+value);
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG,"对error事件做出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG,"对Complete事件做出响应");
            }
        });

    }

log信息：

06-14 11:03:27.138 7493-7493/com.gjj.frame D/DeferActivity: 开始采用subscribe连接
06-14 11:03:27.138 7493-7493/com.gjj.frame D/DeferActivity: 接收到的整数是15
06-14 11:03:27.138 7493-7493/com.gjj.frame D/DeferActivity: 对Complete事件做出响应

9. timer()

延迟指定时间后，发送一个数值0(Long类型)，一般用于检测

        Observable.timer(4, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(Constant.TAG,"开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Long value) {
                        Log.d(Constant.TAG,"接收到的整数是"+value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(Constant.TAG,"对error事件做出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(Constant.TAG,"对Complete事件做出响应");
                    }
                });

log信息：

06-14 11:11:19.802 8110-8110/com.gjj.frame D/RxJava: 开始采用subscribe连接
06-14 11:11:23.805 8110-8151/com.gjj.frame D/RxJava: 接收到的整数是0
06-14 11:11:23.806 8110-8151/com.gjj.frame D/RxJava: 对Complete事件做出响应

10. interval()

每隔指定时间就发送事件，发送事件的序列从0开始无限递增1的整数序列。

       //参数说明：
      //参数1：第一次延迟时间
      //参数2：第二次延迟数字
      //参数3：时间单位
        Observable.interval(3,1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(Constant.TAG,"开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Long value) {
                        Log.d(Constant.TAG,"接收到了事件"+value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(Constant.TAG,"对error事件做出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(Constant.TAG,"对Complete事件做出响应");
                    }
                });

11. intervalRange()

每隔指定时间就发送事件，可指定发送的数据的数量，发送的事件序列从0开始、无限递增1的整数序列

        //参数说明：
        //参数1：事件序列起始点；
        //参数2：事件数量
        //参数3：第1次事件延迟发送时间
        //参数4：时间间隔
        //参数5：参数单位
        Observable.intervalRange(3,10,2,1, TimeUnit.SECONDS)
                //从3开始一共发送10个数据
                //第一次延迟2s发送，之后每个1s产生一个数字（从0开始递增1）
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(Constant.TAG,"开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Long value) {
                        Log.d(Constant.TAG,"接收到了事件"+value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(Constant.TAG,"对error事件做出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(Constant.TAG,"对Complete事件做出响应");
                    }
                });

log信息：

06-14 11:33:05.721 9141-9141/com.gjj.frame D/RxJava: 开始采用subscribe连接
06-14 11:33:07.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件3
06-14 11:33:08.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件4
06-14 11:33:09.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件5
06-14 11:33:10.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件6
06-14 11:33:11.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件7
06-14 11:33:12.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件8
06-14 11:33:13.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件9
06-14 11:33:14.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件10
06-14 11:33:15.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件11
06-14 11:33:16.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件12
06-14 11:33:16.727 9141-9238/com.gjj.frame D/RxJava: 对Complete事件做出响应

12. range()

连续发送一个事件，可指定范围

        //参数说明：
        //参数1：事件序列起始点
        //参数2：事件数量
        Observable.range(3,10)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(Constant.TAG,"开始采用subscribe连接");
                    }

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

                    @Override
                    public void onError(Throwable e) {
                        Log.e(Constant.TAG,"对error事件做出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(Constant.TAG,"对Complete事件做出响应");
                    }
                });

log信息：

06-14 12:12:55.184 10356-10356/com.gjj.frame D/RxJava: 开始采用subscribe连接
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件3
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava:  接收到了事件4
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件5
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件6
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件7
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 对Complete事件做出响应

13. rangeLong()

类似于rang,区别在于该方法支持数据类型long

参考文章：

Android RxJava：最基础的操作符详解 - 创建操作符

RxJava操作符-->创建

引言

1. create()

2.just()

3. fromArray()

4. fromIterable()

5. never()

6. empty()

7. error()

8. defer()

9. timer()

10. interval()

11. intervalRange()

12. range()

13. rangeLong()

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读