RxJava- 操作符之组合Observable

Posted 2020-07-07 行云间

      

       在实际开发过程中，实现一个功能，需要通过多个线程来处理。比如，房间温控监视仪需要监视不同房间的温度，温度传感器会实时传递并保存至数据库，意味着我们需要处理多个Observables。在这种情况下，该如何处理这些Observables呢？本篇博客，我们将学习组合操作符，了解如何同时处理多个Observables来创建我们想要的Observable。

merge

    merge操作符将两个甚至更多的Observables合并到一个发射的数据项里，并发射出去。
    merge操作符的流程图如下：

    示例代码:

        List<Student> list_0 = new ArrayList<>();
        List<Student> list_1 = new ArrayList<>();
        list_0.add(new Student("Merge-A11", "20", "1101"));
        list_0.add(new Student("Merge-A12", "23", "1102"));
        list_0.add(new Student("Merge-A13", "22", "1103"));
        list_0.add(new Student("Merge-A14", "21", "1104"));
        list_0.add(new Student("Merge-A15", "20", "1105"));

        list_1.add(new Student("Merge-B11", "20", "1101"));
        list_1.add(new Student("Merge-B12", "23", "1102"));
        list_1.add(new Student("Merge-B13", "22", "1103"));

        Observable<Student> obs_0 = Observable.from(list_0);
        Observable<Student> obs_1 = Observable.from(list_1);
        Observable<Student> obsMerge = Observable.merge(obs_0, obs_1);
        obsMerge.subscribe(new Observer<Student>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Student student) {
                        mAdaStu.addData(student);
                    }
                });

    
       如示例代码，我们创建了两个Observable数据列表obs_0和obs_1，使用Observable.merge()，将obs_0和obs_1合并，并创建新的Observables列表 obsMerge，它在单个可观测序列中发射源Observables发出的所有数据。
    
       注: 合并后的新Observables发射时，可以认为任何一个Observable发生错误，都将会打断合并。如果想避免这种情况发生，可以调用mergeDelayError()方法，表明它能从一个Observable中继续发射数据即便是其中有一个抛出了错误。当所有的Observables都完成时，mergeDelayError()将会发射onError()。
   

zip

    zip操作符合并两个或者多个Observables发射出的数据项，根据指定的函数Func*变换它们，并发射一个新值。
    zip操作符的流程图如下：
    
    示例代码:

    Observable<Student> obs_stu = Observable.from(mLists);
    Observable<Long> obs_long = Observable.interval(1, TimeUnit.SECONDS);
    Observable.zip(obs_stu, obs_long, new Func2<Student, Long, Student>() {
            @Override
            public Student call(Student student, Long aLong) {
                return updateTitle(student, aLong);
            }
        }).observeOn(androidSchedulers.mainThread())
         .subscribe(new Observer<Student>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Student student) {
                mAdaStu.addData(student);
            }
        });
     
      private Student updateTitle(Student stu, Long time) {
        stu.setName(time + " - " + stu.getName());
        return stu;
    }

    
       从zip操作符流程图和示例代码，可以明显看出zip将Observables列表obs_stu和obs_long数据，在函数Func里通过调用
updateTitle(Student stu, Long time)方法，将两个列表的数值一一对应的合并并发射。
       zip函数有三个关键的参数：两个Observables和一个Func。其实，两个Observables就是Observable的数据源，而Func方法用于将两个Observables在相对应的规则下合并并发射。

join

       join操作符是也是两个Observable产生的结果进行合并，合并的结果组成一个新的Observable，但是join操作符可以控制每个Observable产生结果的生命周期，在每个结果的生命周期内，可以与另一个Observable产生的结果按照一定的规则进行合并。
       join操作符流程图:
    
    示例代码:

    Observable<Student> obs_stu = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, Student>() {
            @Override
            public Student call(Long aLong) {
                return mLists.get(aLong.intValue());
            }
        });
    Observable<Long> obs_time = Observable.interval(1, TimeUnit.SECONDS);

    obs_stu.join(obs_time,
            student -> Observable.timer(2,TimeUnit.SECONDS),
            time -> Observable.interval(2, TimeUnit.SECONDS),
            this::updateTitle)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Student>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Student student) {
                    mAdaStu.addData(student);

                    mListsAdd.add(student);
                    int position = mListsAdd.size() - 1;
                    mAdaStu.addDataByPosition(position, student);
                    rvCombining.smoothScrollToPosition(position);
                }
            });

    从上述代码可以看出，join的过程：
    1.定义数据源：源Observable序列obs_stu将每隔1s发射一个数据的数据源，而obs_time是每隔1s发射一个新的Long整型。
    2.整合数据：为了整合它们发的数据，通过
    student -> Observable.timer(2,TimeUnit.SECONDS)
    time -> Observable.interval(2, TimeUnit.SECONDS)
    限定在一定的生命周期内，将obs_stu发射的数据和obs_time发射的数据相互配合返回的Observable。
    3.合并数据：通过Func2，定义合并规则合并发射
    
    join方法的用法如下：
    observableA.join(observableB,
    observableA产生结果生命周期控制函数，
    observableB产生结果生命周期控制函数，
    observableA产生的结果与observableB产生的结果的合并规则）
    参数解析：
    * observableB：与源Observable结合的第二个Observable
    * observableA产生结果生命周期控制函数:在指定的由生命周期内，源Observable发射的数据和从第二个Observable发射的数据相互配合返回的Observable.
    * observableB产生结果生命周期控制函数:在指定的由生命周期内，第二个Observable发射的数据和从源Observable发射的数据相互配合返回的Observable。
    * 合并规则FUNC：将定义已发射的数据如何与新发射的数据项相结合在相对应的规则下合并并发射。

combineLatest

    combineLatest操作符有点类似zip操作符的特殊形式，用于将最近发射数据。看下方的流程图，如果Observable1发射了2并且Observable2发射了A、B、C、D,combineLatest()将会分组处理2A,2B,2C和2D.Observable1发射了3后，combineLatest()将会处理3D.

    combineLatest操作符流程图:

    示例代码：

    Observable<Student> obs_stu = Observable.interval(1, TimeUnit.SECONDS)
        .map(position -> mLists.get(position.intValue()));
    Observable<Long> obs_time =
            Observable.interval(1500, TimeUnit.MILLISECONDS);

    Observable.combineLatest(obs_stu,
            obs_time,
            this::updateTitle)
            . observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Student>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Student student) {
                    mAdaStu.addData(student);
                }
            });

And,Then和When

      And,Then和When操作符是在RxJava的joins包下，使用Pattern和Plan作为中介，将发射的数据集合并到一起。需要导包joins包，AndroidStudio中在gradle引用：compile 'io.reactivex:rxjava-joins:0.22.0'，如果不引用此包，类Pattern、Plan和JoinObservable不存在.

    And,Then和When操作符的流程图如下：

    示例代码：

    Observable<Student> obs_stu = Observable.from(mLists);
    Observable<Long> obs_time = Observable.interval(1, TimeUnit.SECONDS);

    Pattern2<Student, Long> pattern = JoinObservable.from(obs_stu).and(obs_time);
    Plan0<Student> plan = pattern.then(this::updateTitle);

    JoinObservable
            .when(plan)
            .toObservable()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Student>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Student student) {
                    mAdaStu.addData(student);
                }
            });

    细看示例代码，And、Then和When流程如下：
    1.创建发射源：obs_stu发射Student数据，obs_time每秒发射一个Long型整数
    2.创建Pattern对象
    3.使用刚创建的Pattern对象，创建一个Plan对象，即我们有两个发射数据的Observables,then()是做什么的？"
    4.当Plan对象存在，如果Plan发生时应该做哪些事情，即.when(plan).toObservable()
    5.最后就是订阅
    

switch

        switch操作符正如定义的，将一个发射多个Observables的Observable转换成另一个单独的Observable，后者发射那些Observables最近发射的数据项。
       给出一个发射多个Observables序列的源Observable，switch()订阅到源Observable然后开始发射由第一个发射的Observable发射的一样的数据。当源Observable发射一个新的Observable时，switch()立即取消订阅前一个发射数据的Observable（因此打断了从它那里发射的数据流）然后订阅一个新的Observable，并开始发射它的数据。
       switch操作符的流程图如下：
    

startWith

    startWith操作符通过传递一个参数来先发射一个数据序列。
    startWith操作符的流程图如下：
    
    示例代码：

    Subscription sub = Observable.interval(1, TimeUnit.SECONDS)
        .map(position -> mLists.get(position.intValue()))
        .startWith(mLists.get(2))
        . observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<Student>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Student student) {
                mAdaStu.addData(student);
            }
        });

    

    正如示例代码中，调用.startWith(mLists.get(2))函数，欲将第三个数据优先从数列中发出。而在实际效果图上，数据A13优先被观察者接收，恰恰印证了这一点。