Android之Rxjava2.X 2————Rxjava 创建操作符

柔情只为你懂 2022-05-16 01:16 307阅读 0赞

# Android之Rxjava2.X 2————Rxjava 创建操作符 #

### 一.目录 ###

### 文章目录 ###

*  Android之Rxjava2.X 2————Rxjava 创建操作符
 *   *   *  一.目录
         *  二.概述
         *   *  1.作用
             *  2. 类型
         *  三.基本创建
         *   *  1. create（）
         *  四.快速创建
         *   *  1.just()
             *  2.fromArray（）
             *  3.fromIterable（）
             *  4.其他
         *  五. 延迟创建
         *   *  1.defer（）
             *  2.timer（）
             *  3.interval（）
             *  4.intervalRange（）
             *  5.range（） /rangeLong（）
         *  六. 总结
         *  七.参考资料
         *  八.文章索引

### 二.概述 ###

#### 1.作用 ####

创建 被观察者（ Observable） 对象 & 发送事件。

#### 2. 类型 ####

![这里写图片描述][61075a6e9d3b25a1abf2adf293504f03.png]

### 三.基本创建 ###

需求场景: 完整的创建被观察者对象

#### 1. create（） ####

你可以使用Create操作符创建一个完整的Observable，可以传递onNext，onError和onCompleted等事件。  
![这里写图片描述][4d6b83876b3b4e40546393bbdc458ea9.png]  
代码示例：

Observable.create(new ObservableOnSubscribe<Integer>() {
        
                @Override
                public void subscribe(ObservableEmitter<Integer> e) throws Exception {
        
                    e.onNext(1);
                    e.onNext(2);
                    e.onNext(3);
                }
            }).subscribe(new Observer<Integer>() {
        
                @Override
                public void onSubscribe(Disposable d) {
        
                    Log.d(TAG, "onSubscribe: ");
                }
    
                @Override
                public void onNext(Integer integer) {
        
                    Log.d(TAG, "onNext: "+integer);
                }
    
                @Override
                public void onError(Throwable e) {
        
                    Log.d(TAG, "onError: "+e);
                }
    
                @Override
                public void onComplete() {
        
                    Log.d(TAG, "onComplete: ");
                }
            });

运行结果  
![这里写图片描述][20180813172532425]

注意:当观察者发送一个Complete/Error事件后，被观察者在，Complrte/Error事件将会继续发送，但观察者收到Complete/Error事件后，不会继续接收任何事件。被观察者可以不发生Complete/Erroe事件

### 四.快速创建 ###

需求场景：快速的创建被观察者对象

#### 1.just() ####

*  作用:快速创建1个被观察者对象（Observable）
 *  发送事件的特点：直接发送传入的事件
 *  注意1:just最多只能发送9个参数
 *  注意2：如果你传递null给Just，它会返回一个发射null值的Observable  
    ![这里写图片描述][8d74efa06567e1066acef6881846cec8.png]  
    代码示例

Observable.just(1, 2, 3)
                   .subscribe(new Observer<Integer>() {
        
                       @Override
                       public void onSubscribe(Disposable d) {
        
                           Log.d(TAG, "onSubscribe: "+d);
                       }
    
                       @Override
                       public void onNext(Integer integer) {
        
                           Log.d(TAG, "onNext: "+integer);
                       }
    
                       @Override
                       public void onError(Throwable e) {
        
                           Log.d(TAG, "onError: ");
                       }
    
                       @Override
                       public void onComplete() {
        
                           Log.d(TAG, "onComplete: ");
                       }
                   });

![这里写图片描述][2018081320191198]

#### 2.fromArray（） ####

*  作用：快速创建一个被观察者对象
 *  发送事件的特点：直接发送传入的数据数组
 *  fromArray会将数组中的数据转换为Observable对象
 *  应用场景:被观察者对象（Observable） & 发送10个以上事件（数组形式),数组遍历

图中的From包括fromArray（）以及下面的fromIterable（）  
![这里写图片描述][bb192d2b2d628fdb0228082294e17fe4.png]

Integer[] items = {
         0, 1, 2, 3, 4 };
            Observable.fromArray(items)
                   .subscribe(new Observer<Integer>() {
        
                       @Override
                       public void onSubscribe(Disposable d) {
        
                           Log.d(TAG, "onSubscribe: "+d);
                       }
    
                       @Override
                       public void onNext(Integer integer) {
        
                           Log.d(TAG, "onNext: "+integer);
                       }
    
                       @Override
                       public void onError(Throwable e) {
        
                           Log.d(TAG, "onError: ");
                       }
    
                       @Override
                       public void onComplete() {
        
                           Log.d(TAG, "onComplete: ");
                       }
                   });

![这里写图片描述][20180813203203119]

#### 3.fromIterable（） ####

*  作用:快速创建1个被观察者对象（Observable）
 *  发送事件的特点:直接发送 传入的集合List数据
 *  应用场景:1.快速创建 被观察者对象（Observable） & 发送10个以上事件（集合形式）2.集合元素遍历

List<Integer> list = new ArrayList<>();
            list.add(1);
            list.add(2);
            list.add(3);
            Observable.fromIterable(list)
                   .subscribe(new Observer<Integer>() {
        
                       @Override
                       public void onSubscribe(Disposable d) {
        
                           Log.d(TAG, "onSubscribe: "+d);
                       }
    
                       @Override
                       public void onNext(Integer integer) {
        
                           Log.d(TAG, "onNext: "+integer);
                       }
    
                       @Override
                       public void onError(Throwable e) {
        
                           Log.d(TAG, "onError: ");
                       }
    
                       @Override
                       public void onComplete() {
        
                           Log.d(TAG, "onComplete: ");
                       }
                   });
        }

![这里写图片描述][20180813203848779]

#### 4.其他 ####

下列方法一般用于测试使用  
empty():仅发送Complete事件，直接通知完成  
error():仅发送Error事件，直接通知异常  
never():不发送任何事件

Observable.empty()
                   .subscribe(new Observer<Object>() {
        
                       @Override
                       public void onSubscribe(Disposable d) {
        
                           Log.d(TAG, "onSubscribe: "+d);
                       }
    
                       @Override
                       public void onNext(Object o) {
        
                           Log.d(TAG, "onNext: "+o);
                       }
    
                       @Override
                       public void onError(Throwable e) {
        
                           Log.d(TAG, "onError: ");
                       }
    
                       @Override
                       public void onComplete() {
        
                           Log.d(TAG, "onComplete: ");
                       }
                   });
    
     Observable.error(new RuntimeException())
                   .subscribe(new Observer<Object>() {
        
                       @Override
                       public void onSubscribe(Disposable d) {
        
                           Log.d(TAG, "onSubscribe: "+d);
                       }
    
                       @Override
                       public void onNext(Object o) {
        
                           Log.d(TAG, "onNext: "+o);
                       }
    
                       @Override
                       public void onError(Throwable e) {
        
                           Log.d(TAG, "onError: ");
                       }
    
                       @Override
                       public void onComplete() {
        
                           Log.d(TAG, "onComplete: ");
                       }
                   });
    Observable.never()
                   .subscribe(new Observer<Object>() {
        
                       @Override
                       public void onSubscribe(Disposable d) {
        
                           Log.d(TAG, "onSubscribe: "+d);
                       }
    
                       @Override
                       public void onNext(Object o) {
        
                           Log.d(TAG, "onNext: "+o);
                       }
    
                       @Override
                       public void onError(Throwable e) {
        
                           Log.d(TAG, "onError: ");
                       }
    
                       @Override
                       public void onComplete() {
        
                           Log.d(TAG, "onComplete: ");
                       }
                   });

![这里写图片描述][2018081320460146]

![这里写图片描述][20180813204946557]

![这里写图片描述][20180813205122569]

### 五. 延迟创建 ###

需求场景：

*  定时操作：在经过了x秒后，需要自动执行y操作
 *  周期性操作：每隔x秒后，需要自动执行y操作

#### 1.defer（） ####

*  作用:defer（）操作符会一直等待直到有观察者订阅它，然后它使用Observable工厂方法生成一个Observable。在某些情况下，等待直到最后一分钟（就是知道订阅发生时）才生成Observable可以确保Observable包含最新的数据。
 *  使用场景：动态创建被观察者对象（Observable） & 获取最新的Observable对象数据

![这里写图片描述][d63424dce829b1fcae8c8d336463b121.png]

//  1. 第1次对i赋值
             i= 10;
    
            Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
        
                @Override
                public ObservableSource<? extends Integer> call() throws Exception {
        
                    return Observable.just(i);
                }
            });
    
            //  1. 第1次对i赋值
            i = 15;
            observable
                    .subscribe(new Observer<Object>() {
        
                        @Override
                        public void onSubscribe(Disposable d) {
        
                            Log.d(TAG, "onSubscribe: " + d);
                        }
    
                        @Override
                        public void onNext(Object o) {
        
                            Log.d(TAG, "onNext: " + o);
                        }
    
                        @Override
                        public void onError(Throwable e) {
        
                            Log.d(TAG, "onError: ");
                        }
    
                        @Override
                        public void onComplete() {
        
                            Log.d(TAG, "onComplete: ");
                        }
                    });

![这里写图片描述][20180813211830829]

#### 2.timer（） ####

*  作用:创建一个Observable，它在一个给定的延迟后发射一个特殊的值。
 *  应用:延迟指定事件，发送一个0，一般用于检测  
    ![这里写图片描述][0f65f6fc7cae7a2cc206ac3a00bd9a3d.png]

// 该例子 = 延迟2s后，发送一个long类型数值
            Observable.timer(2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
        
                @Override
                public void onSubscribe(Disposable d) {
        
                    Log.d(TAG, "开始采用subscribe连接");
                }
    
                @Override
                public void onNext(Long value) {
        
                    Log.d(TAG, "接收到了事件" + value);
                }
    
                @Override
                public void onError(Throwable e) {
        
                    Log.d(TAG, "对Error事件作出响应");
                }
    
                @Override
                public void onComplete() {
        
                    Log.d(TAG, "对Complete事件作出响应");
                }
            });

![这里写图片描述][20180813212414407]

#### 3.interval（） ####

*  作用 : 按固定的时间间隔发射一个无限递增的整数序列。
 *  interval(long,TimeUnit,Scheduler))
 *  参数说明： 参数1 = 第1次延迟时间，参数2 = 间隔时间数字，参数3 = 时间单位；  
    ![这里写图片描述][20180813213614682]

Observable.interval(3, 1, TimeUnit.SECONDS)
                    // 该例子发送的事件序列特点：延迟3s后发送事件，每隔1秒产生1个数字（从0开始递增1，无限个）
                    .subscribe(new Observer<Long>() {
        
                        @Override
                        public void onSubscribe(Disposable d) {
        
                            Log.d(TAG, "开始采用subscribe连接");
                        }
    
                        @Override
                        public void onNext(Long value) {
        
                            Log.d(TAG, "接收到了事件" + value);
                        }
    
                        @Override
                        public void onError(Throwable e) {
        
                            Log.d(TAG, "对Error事件作出响应");
                        }
    
                        @Override
                        public void onComplete() {
        
                            Log.d(TAG, "对Complete事件作出响应");
                        }
                    });

![这里写图片描述][20180813213121499]

#### 4.intervalRange（） ####

*  作用:发送事件的特点：每隔指定时间 就发送 事件，可指定发送的数据（从0开始、无限递增1的的整数）的数量

// 参数1 = 事件序列起始点；
            // 参数2 = 事件数量；
            // 参数3 = 第1次事件延迟发送时间；
            // 参数4 = 间隔时间数字；
            // 参数5 = 时间单位
    Observable.intervalRange(3, 10, 2, 1, TimeUnit.SECONDS)
                    // 该例子发送的事件序列特点： 
                    // 1. 从3开始，一共发送10个事件； 
                    // 2. 第1次延迟2s发送，之后每隔2秒产生1个数字（从0开始递增1，无限个） 
                    .subscribe(new Observer<Long>() {
        
                        @Override
                        public void onSubscribe(Disposable d) {
        
                            Log.d(TAG, "开始采用subscribe连接");
                        } // 默认最先调用复写的 onSubscribe（） 
    
                        @Override
                        public void onNext(Long value) {
        
                            Log.d(TAG, "接收到了事件" + value);
                        }
    
                        @Override
                        public void onError(Throwable e) {
        
                            Log.d(TAG, "对Error事件作出响应");
                        }
    
                        @Override
                        public void onComplete() {
        
                            Log.d(TAG, "对Complete事件作出响应");
                        }
                    });

![这里写图片描述][2018081321393671]

#### 5.range（） /rangeLong（） ####

range（） 作用：连续发送 1个事件序列，可指定范围,作用类似于intervalRange（），但区别在于：无延迟发送事件

rangeLong（）类似于range（），区别在于该方法支持数据类型 = Long  
![这里写图片描述][e978786556bdc0e929e9d9ebf5796d81.png]

// 参数说明：
            // 参数1 = 事件序列起始点；
            // 参数2 = 事件数量；
            // 注：若设置为负数，则会抛出异常
            Observable.range(3,10)
                    // 该例子发送的事件序列特点：从3开始发送，每次发送事件递增1，一共发送10个事件
                    .subscribe(new Observer<Integer>() {
        
                        @Override
                        public void onSubscribe(Disposable d) {
        
                            Log.d(TAG, "开始采用subscribe连接");
                        } // 默认最先调用复写的 onSubscribe（）
    
                        @Override
                        public void onNext(Integer value) {
        
                            Log.d(TAG, "接收到了事件" + value);
                        }
    
                        @Override
                        public void onError(Throwable e) {
        
                            Log.d(TAG, "对Error事件作出响应");
                        }
    
                        @Override
                        public void onComplete() {
        
                            Log.d(TAG, "对Complete事件作出响应");
                        }
                    });

![这里写图片描述][20180813215515119]

### 六. 总结 ###

![这里写图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NDk5ODU5_size_16_color_FFFFFF_t_70]

### 七.参考资料 ###

[ReactiveX文档中文翻译][ReactiveX]  
[Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程 ][Android Rxjava_ _ _ _Rxjava _]  
Rxjava2入门教程一：函数响应式编程及概述  
[这可能是最好的RxJava 2.x 教程（完结版）][RxJava 2.x]  
[那些年我们错过的响应式编程][Link 1]

### 八.文章索引 ###

*  [Android之Rxjava2.X 1————Rxjava概述][Android_Rxjava2.X 1_Rxjava]
 *  [Android之Rxjava2.X 2————Rxjava 创建操作符][Android_Rxjava2.X 2_Rxjava]
 *  [Android之Rxjava2.X 3————Rxjava 变换操作符][Android_Rxjava2.X 3_Rxjava]
 *  [Android之Rxjava2.X 4————Rxjava 组合操作符][Android_Rxjava2.X 4_Rxjava]
 *  [Android之Rxjava2.X 5————Rxjava 过滤操作符][Android_Rxjava2.X 5_Rxjava]
 *  [Android之Rxjava2.X 6————Rxjava 功能操作符][Android_Rxjava2.X 6_Rxjava]
 *  [Android之Rxjava2.X 7————Rxjava 条件操作符][Android_Rxjava2.X 7_Rxjava]
 *  [Android之Rxjava2.X 8————Rxjava 背压策略][Android_Rxjava2.X 8_Rxjava]
 *  [Android之Rxjava2.X 9————Rxjava源码阅读1][Android_Rxjava2.X 9_Rxjava_1]
 *  [Android之Rxjava2.X 10————Rxjava源码阅读2][Android_Rxjava2.X 10_Rxjava_2]
 *  [Android之Rxjava2.X 11————Rxjava源码阅读3][Android_Rxjava2.X 11_Rxjava_3]

[61075a6e9d3b25a1abf2adf293504f03.png]: /images/20220516/b45c882e7872407caeb992745049df70.png
[4d6b83876b3b4e40546393bbdc458ea9.png]: /images/20220516/5a637af7e12a4817a0a7a97a2b072692.png
[20180813172532425]: /images/20220516/af5eb69b70614d60a5f1f181fb73e36a.png
[8d74efa06567e1066acef6881846cec8.png]: /images/20220516/073cb6844ea646cfb583f0cdf1c7a25b.png
[2018081320191198]: /images/20220516/91a2f0ac9c3c425da44f1a4468790316.png
[bb192d2b2d628fdb0228082294e17fe4.png]: /images/20220516/e906c8af4cbb4d74b748f28b75f215f0.png
[20180813203203119]: /images/20220516/1c9913d755134f3e8d809d10c6260cf8.png
[20180813203848779]: /images/20220516/68774c85aeb544d3a724e2d9e0445aff.png
[2018081320460146]: /images/20220516/5f79107e9e0b45b1b4c75cafe2ab1654.png
[20180813204946557]: /images/20220516/5b66fc6624e549ed9611ce1dc4f62505.png
[20180813205122569]: /images/20220516/3c9b2762583840a6a54c6dd1cad7d563.png
[d63424dce829b1fcae8c8d336463b121.png]: /images/20220516/581964cfc7364ca59c02c859f8e1bbd9.png
[20180813211830829]: /images/20220516/76ffc9a5568f44b29ce12c8894045d54.png
[0f65f6fc7cae7a2cc206ac3a00bd9a3d.png]: /images/20220516/ac0c6d877758415ea0909bd04f2f32b9.png
[20180813212414407]: /images/20220516/f1dc4b072381484e837cecff44eda86b.png
[20180813213614682]: /images/20220516/e37452db6ad141f480b8ade0f5a4adda.png
[20180813213121499]: /images/20220516/7206c049e21741438c6a55f97de7ec85.png
[2018081321393671]: /images/20220516/ad0f39390b6b4de28ca3d10d7b966ed8.png
[e978786556bdc0e929e9d9ebf5796d81.png]: /images/20220516/09f07b0e80ca472e988beae6450209ea.png
[20180813215515119]: /images/20220516/9c1184de0fec479e9ccc797f414c50d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NDk5ODU5_size_16_color_FFFFFF_t_70]: /images/20220516/a94c6617554740839e0177c154494745.png
[ReactiveX]: https://mcxiaoke.gitbooks.io/rxdocs/content/topics/Getting-Started.html
[Android Rxjava_ _ _ _Rxjava _]: https://www.jianshu.com/p/a406b94f3188
[RxJava 2.x]: https://www.jianshu.com/p/0cd258eecf60
[Link 1]: https://github.com/kevinyaoo/android-tech-frontier/tree/master/androidweekly/%E9%82%A3%E4%BA%9B%E5%B9%B4%E6%88%91%E4%BB%AC%E9%94%99%E8%BF%87%E7%9A%84%E5%93%8D%E5%BA%94%E5%BC%8F%E7%BC%96%E7%A8%8B
[Android_Rxjava2.X 1_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81611870
[Android_Rxjava2.X 2_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81637932
[Android_Rxjava2.X 3_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81668545
[Android_Rxjava2.X 4_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81670980
[Android_Rxjava2.X 5_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81675322
[Android_Rxjava2.X 6_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81700187
[Android_Rxjava2.X 7_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81705745
[Android_Rxjava2.X 8_Rxjava]: https://blog.csdn.net/qq_38499859/article/details/81747334
[Android_Rxjava2.X 9_Rxjava_1]: https://blog.csdn.net/qq_38499859/article/details/81775520
[Android_Rxjava2.X 10_Rxjava_2]: https://blog.csdn.net/qq_38499859/article/details/81839955
[Android_Rxjava2.X 11_Rxjava_3]: https://blog.csdn.net/qq_38499859/article/details/82119900