前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Android RxJava操作符详解 系列:组合 / 合并操作符

Android RxJava操作符详解 系列:组合 / 合并操作符

作者头像
Carson.Ho
发布2019-02-22 14:58:52
2.1K0
发布2019-02-22 14:58:52
举报
文章被收录于专栏:Android知识分享

前言

  • Rxjava,由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点,深受各大 Android开发者的欢迎。
Github截图
Github截图

如果还不了解 RxJava,请看文章:Android:这是一篇 清晰 & 易懂的Rxjava 入门教程

  • RxJava如此受欢迎的原因,在于其提供了丰富 & 功能强大的操作符,几乎能完成所有的功能需求
  • 今天,我将为大家详细介绍RxJava操作符中最常用的 组合 / 合并操作符,并附带 Retrofit 结合 RxJava的实例Demo教学,希望你们会喜欢。
    1. 本系列文章主要基于 Rxjava 2.0
    2. 接下来的时间,我将持续推出 AndroidRxjava 2.0 的一系列文章,包括原理、操作符、应用场景、背压等等 ,有兴趣可以继续关注Carson_Ho的安卓开发笔记!!
示意图
示意图

目录

示意图
示意图

1. 作用

组合 多个被观察者(`Observable`) & 合并需要发送的事件


2. 类型

  • RxJava 2 中,常见的组合 / 合并操作符 主要有:
示意图
示意图
  • 下面,我将对每个操作符进行详细讲解

3. 应用场景 & 对应操作符 介绍

注:在使用RxJava 2操作符前,记得在项目的Gradle中添加依赖:

代码语言:javascript
复制
dependencies {
      compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
      compile 'io.reactivex.rxjava2:rxjava:2.0.7'
      // 注:RxJava2 与 RxJava1 不能共存,即依赖不能同时存在
}

3.1 组合多个被观察者

该类型的操作符的作用 = 组合多个被观察者

concat() / concatArray()

  • 作用 组合多个被观察者一起发送数据,合并后 按发送顺序串行执行 二者区别:组合被观察者的数量,即concat()组合被观察者数量≤4个,而concatArray()则可>4个
  • 具体使用
代码语言:javascript
复制
// concat():组合多个被观察者(≤4个)一起发送数据
        // 注:串行执行
        Observable.concat(Observable.just(1, 2, 3),
                           Observable.just(4, 5, 6),
                           Observable.just(7, 8, 9),
                           Observable.just(10, 11, 12))
                  .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @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事件作出响应");
                    }
                });

// concatArray():组合多个被观察者一起发送数据(可>4个)
        // 注:串行执行
        Observable.concatArray(Observable.just(1, 2, 3),
                           Observable.just(4, 5, 6),
                           Observable.just(7, 8, 9),
                           Observable.just(10, 11, 12),
                           Observable.just(13, 14, 15))
                  .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @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事件作出响应");
                    }
                });
  • 测试结果
concat()情况
concat()情况
concatArray()情况
concatArray()情况

merge() / mergeArray()

  • 作用 组合多个被观察者一起发送数据,合并后 按时间线并行执行
    1. 二者区别:组合被观察者的数量,即merge()组合被观察者数量≤4个,而mergeArray()则可>4个
    2. 区别上述concat()操作符:同样是组合多个被观察者一起发送数据,但concat()操作符合并后是按发送顺序串行执行
  • 具体使用
代码语言:javascript
复制
// merge():组合多个被观察者(<4个)一起发送数据
        // 注:合并后按照时间线并行执行
        Observable.merge(
                Observable.intervalRange(0, 3, 1, 1, TimeUnit.SECONDS), // 从0开始发送、共发送3个数据、第1次事件延迟发送时间 = 1s、间隔时间 = 1s
                Observable.intervalRange(2, 3, 1, 1, TimeUnit.SECONDS)) // 从2开始发送、共发送3个数据、第1次事件延迟发送时间 = 1s、间隔时间 = 1s
                  .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @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事件作出响应");
                    }
                });

// mergeArray() = 组合4个以上的被观察者一起发送数据,此处不作过多演示,类似concatArray()
  • 测试结果

两个被观察者发送事件并行执行,输出结果 = 0,2 -> 1,3 -> 2,4

33333.gif
33333.gif

concatDelayError() / mergeDelayError()

  • 作用
示意图
示意图
  • 具体使用

a. 无使用concatDelayError()的情况

代码语言:javascript
复制
Observable.concat(
                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 NullPointerException()); // 发送Error事件,因为无使用concatDelayError,所以第2个Observable将不会发送事件
                        emitter.onComplete();
                    }
                }),
                Observable.just(4, 5, 6))
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }
                    @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事件作出响应");
                    }
                });

测试结果:第1个被观察者发送Error事件后,第2个被观察者则不会继续发送事件

示意图
示意图
代码语言:javascript
复制
<-- 使用了concatDelayError()的情况 -->
Observable.concatArrayDelayError(
                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 NullPointerException()); // 发送Error事件,因为使用了concatDelayError,所以第2个Observable将会发送事件,等发送完毕后,再发送错误事件
                        emitter.onComplete();
                    }
                }),
                Observable.just(4, 5, 6))
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }
                    @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事件作出响应");
                    }
                });

测试结果:第1个被观察者的Error事件将在第2个被观察者发送完事件后再继续发送

示意图
示意图

mergeDelayError()操作符同理,此处不作过多演示


3.2 合并多个事件

该类型的操作符主要是对多个被观察者中的事件进行合并处理。

Zip()

  • 作用 合并 多个被观察者(Observable)发送的事件,生成一个新的事件序列(即组合过后的事件序列),并最终发送
  • 原理 具体请看下图
示意图
示意图
  • 特别注意:
    1. 事件组合方式 = 严格按照原先事件序列 进行对位合并
    2. 最终合并的事件数量 = 多个被观察者(Observable)中数量最少的数量 即如下图
示意图
示意图
  • 具体使用
代码语言:javascript
复制
<-- 创建第1个被观察者 -->
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "被观察者1发送了事件1");
                emitter.onNext(1);
                // 为了方便展示效果,所以在发送事件后加入2s的延迟
                Thread.sleep(1000);

                Log.d(TAG, "被观察者1发送了事件2");
                emitter.onNext(2);
                Thread.sleep(1000);

                Log.d(TAG, "被观察者1发送了事件3");
                emitter.onNext(3);
                Thread.sleep(1000);

                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io()); // 设置被观察者1在工作线程1中工作

<-- 创建第2个被观察者 -->
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "被观察者2发送了事件A");
                emitter.onNext("A");
                Thread.sleep(1000);

                Log.d(TAG, "被观察者2发送了事件B");
                emitter.onNext("B");
                Thread.sleep(1000);

                Log.d(TAG, "被观察者2发送了事件C");
                emitter.onNext("C");
                Thread.sleep(1000);

                Log.d(TAG, "被观察者2发送了事件D");
                emitter.onNext("D");
                Thread.sleep(1000);

                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.newThread());// 设置被观察者2在工作线程2中工作
        // 假设不作线程控制,则该两个被观察者会在同一个线程中工作,即发送事件存在先后顺序,而不是同时发送

<-- 使用zip变换操作符进行事件合并 -->
// 注:创建BiFunction对象传入的第3个参数 = 合并后数据的数据类型
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String string) throws Exception {
                return  integer + string;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(TAG, "最终接收到的事件 =  " + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        });
  • 测试结果
示意图
示意图
  • 特别注意:
    1. 尽管被观察者2的事件D没有事件与其合并,但还是会继续发送
    2. 若在被观察者1 & 被观察者2的事件序列最后发送onComplete()事件,则被观察者2的事件D也不会发送,测试结果如下
示意图
示意图
  • 因为Zip()操作符较为复杂 & 难理解,此处将用1张图总结
示意图
示意图

关于Zip()结合RxJavaRxtrofit的实例讲解将在第4节中详细讲解


combineLatest()

  • 作用 当两个Observables中的任何一个发送了数据后,将先发送了数据的Observables 的最新(最后)一个数据 与 另外一个Observable发送的每个数据结合,最终基于该函数的结果发送数据 与Zip()的区别:Zip() = 按个数合并,即1对1合并;CombineLatest() = 按时间合并,即在同一个时间点上合并
  • 具体使用
代码语言:javascript
复制
Observable.combineLatest(
                    Observable.just(1L, 2L, 3L), // 第1个发送数据事件的Observable
                    Observable.intervalRange(0, 3, 1, 1, TimeUnit.SECONDS), // 第2个发送数据事件的Observable:从0开始发送、共发送3个数据、第1次事件延迟发送时间 = 1s、间隔时间 = 1s
                    new BiFunction<Long, Long, Long>() {
                @Override
                public Long apply(Long o1, Long o2) throws Exception {
                    // o1 = 第1个Observable发送的最新(最后)1个数据
                    // o2 = 第2个Observable发送的每1个数据
                    Log.e(TAG, "合并的数据是: "+ o1 + " "+ o2);
                    return o1 + o2;
                    // 合并的逻辑 = 相加
                    // 即第1个Observable发送的最后1个数据 与 第2个Observable发送的每1个数据进行相加
                }
            }).subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long s) throws Exception {
                    Log.e(TAG, "合并的结果是: "+s);
                }
            });
  • 测试结果
示意图
示意图

combineLatestDelayError()

作用类似于concatDelayError() / mergeDelayError() ,即错误处理,此处不作过多描述

reduce()

  • 作用 把被观察者需要发送的事件聚合成1个事件 & 发送 聚合的逻辑根据需求撰写,但本质都是前2个数据聚合,然后与后1个数据继续进行聚合,依次类推
  • 具体使用
代码语言:javascript
复制
Observable.just(1,2,3,4)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    // 在该复写方法中复写聚合的逻辑
                    @Override
                    public Integer apply(@NonNull Integer s1, @NonNull Integer s2) throws Exception {
                        Log.e(TAG, "本次计算的数据是: "+s1 +" 乘 "+ s2);
                        return s1 * s2;
                        // 本次聚合的逻辑是:全部数据相乘起来
                        // 原理:第1次取前2个数据相乘,之后每次获取到的数据 = 返回的数据x原始下1个数据每
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer s) throws Exception {
                Log.e(TAG, "最终计算的结果是: "+s);

            }
        });
  • 测试结果
示意图
示意图

collect()

  • 作用 将被观察者Observable发送的数据事件收集到一个数据结构里
  • 具体使用
代码语言:javascript
复制
Observable.just(1, 2, 3 ,4, 5, 6)
                .collect(
                        // 1. 创建数据结构(容器),用于收集被观察者发送的数据
                        new Callable<ArrayList<Integer>>() {
                            @Override
                            public ArrayList<Integer> call() throws Exception {
                                return new ArrayList<>();
                            }
                            // 2. 对发送的数据进行收集
                        }, new BiConsumer<ArrayList<Integer>, Integer>() {
                            @Override
                            public void accept(ArrayList<Integer> list, Integer integer)
                                    throws Exception {
                                // 参数说明:list = 容器,integer = 后者数据
                                list.add(integer);
                                // 对发送的数据进行收集
                            }
                        }).subscribe(new Consumer<ArrayList<Integer>>() {
            @Override
            public void accept(@NonNull ArrayList<Integer> s) throws Exception {
                Log.e(TAG, "本次发送的数据是: "+s);

            }
        });
  • 测试结果
示意图
示意图

3.3 发送事件前追加发送事件

startWith() / startWithArray()

  • 作用 在一个被观察者发送事件前,追加发送一些数据 / 一个新的被观察者
  • 具体使用
代码语言:javascript
复制
<-- 在一个被观察者发送事件前,追加发送一些数据 -->
        // 注:追加数据顺序 = 后调用先追加
        Observable.just(4, 5, 6)
                  .startWith(0)  // 追加单个数据 = startWith()
                  .startWithArray(1, 2, 3) // 追加多个数据 = startWithArray()
                  .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @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事件作出响应");
                    }
                });


<-- 在一个被观察者发送事件前,追加发送被观察者 & 发送数据 -->
        // 注:追加数据顺序 = 后调用先追加
        Observable.just(4, 5, 6)
                .startWith(Observable.just(1, 2, 3))
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @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事件作出响应");
                    }
                });
  • 测试结果
示意图
示意图
示意图
示意图

3.4 统计发送事件数量

count()

  • 作用 统计被观察者发送事件的数量
  • 具体使用
代码语言:javascript
复制
// 注:返回结果 = Long类型
        Observable.just(1, 2, 3, 4)
                  .count()
                  .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "发送的事件数量 =  "+aLong);

                    }
                });
  • 测试结果
示意图
示意图

至此,RxJava 2中的组合 / 合并操作符讲解完毕。


4. 实际开发需求案例

下面,我将讲解组合 / 合并操作符的常见实际需求: 1. 从缓存(磁盘、内存)中获取缓存数据 2. 合并数据源 3. 联合判断

  • 下面,我将对每个应用场景进行实例Demo演示讲解。

4.1 获取缓存数据

4.2 合并数据源 & 同时展示

4.3 联合判断

  • 即,同时对多个事件进行联合判断 如,填写表单时,需要表单里所有信息(姓名、年龄、职业等)都被填写后,才允许点击 “提交” 按钮
  • 具体请看文章:Android RxJava 实际应用讲解:联合判断

5. Demo地址

上述所有的Demo源代码都存放在:Carson_Ho的Github地址:RxJava2_组合 / 合并操作符


6. 总结

  • 下面,我将用一张图总结 RxJava2 中常用的组合 / 合并操作符
示意图
示意图
  • 接下来的时间,我将持续推出 AndroidRxjava 2.0 的一系列文章,包括原理、操作符、应用场景、背压等等 ,有兴趣可以继续关注Carson_Ho的安卓开发笔记!!
示意图
示意图

请帮顶 / 评论点赞!因为你的鼓励是我写作的最大动力!

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2017年11月06日,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 前言
  • 目录
  • 1. 作用
  • 2. 类型
  • 3. 应用场景 & 对应操作符 介绍
    • 3.1 组合多个被观察者
      • concat() / concatArray()
        • merge() / mergeArray()
          • concatDelayError() / mergeDelayError()
            • 3.2 合并多个事件
              • Zip()
                • combineLatest()
                  • combineLatestDelayError()
                    • reduce()
                      • collect()
                        • 3.3 发送事件前追加发送事件
                          • startWith() / startWithArray()
                            • 3.4 统计发送事件数量
                              • count()
                              • 4. 实际开发需求案例
                                • 4.1 获取缓存数据
                                  • 4.2 合并数据源 & 同时展示
                                    • 4.3 联合判断
                                    • 5. Demo地址
                                    • 6. 总结
                                    • 请帮顶 / 评论点赞!因为你的鼓励是我写作的最大动力!
                                    相关产品与服务
                                    容器服务
                                    腾讯云容器服务(Tencent Kubernetes Engine, TKE)基于原生 kubernetes 提供以容器为核心的、高度可扩展的高性能容器管理服务,覆盖 Serverless、边缘计算、分布式云等多种业务部署场景,业内首创单个集群兼容多种计算节点的容器资源管理模式。同时产品作为云原生 Finops 领先布道者,主导开源项目Crane,全面助力客户实现资源优化、成本控制。
                                    领券
                                    问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档