RxJava進(jìn)階四（組合類操作符）

RxJava進(jìn)階

RxJava進(jìn)階一（創(chuàng)建類操作符）
RxJava進(jìn)階二（轉(zhuǎn)換類操作符）
RxJava進(jìn)階三（過濾類操作符）
RxJava進(jìn)階四（組合類操作符）

前言

本篇文章帶著大家熟悉一下RxJava的組合類操作符，本系列文章僅是帶大家認(rèn)識(shí)一下這些操作符的用法，并沒有對(duì)操作符進(jìn)行多種形態(tài)的使用，具體的還需要大家在使用時(shí)注意~

操作符總覽

CombineLatest、Join、Merge、StartWith、Switch、Zip...

具體使用介紹

CombineLatest

當(dāng)兩個(gè)Observables中的任何一個(gè)發(fā)射了一個(gè)數(shù)據(jù)時(shí)，將兩個(gè)Observables數(shù)據(jù)通過指定的規(guī)則進(jìn)行處理，將結(jié)果進(jìn)行發(fā)射~

代碼示例：

        Observable<Long> observable1 = Observable.interval(0, 500, TimeUnit.MILLISECONDS).take(3);
        Observable<Long> observable2 = Observable.interval(500, 500, TimeUnit.MILLISECONDS).take(3);

        Observable.combineLatest(observable1, observable2, new Func2<Long, Long, Long>() {
            @Override
            public Long call(Long along1, Long along2) {
                System.out.println("along1 --> " + along1);
                System.out.println("along2 --> " + along2);
                return along1 + along2;
            }
        }).subscribe(new Action1<Long>() {
            @Override
            public void call(Long aLong) {
                System.out.println("result = " + aLong);
                System.out.println("--------------");
            }
        });

運(yùn)行結(jié)果：

10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: along1 --> 1
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: along2 --> 0
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: result = 1
10-08 17:00:12.319 6650-6698/com.shenghan.haobaobei I/System.out: --------------
10-08 17:00:12.816 6650-6728/com.shenghan.haobaobei I/System.out: along1 --> 2
10-08 17:00:12.816 6650-6728/com.shenghan.haobaobei I/System.out: along2 --> 0
10-08 17:00:12.816 6650-6728/com.shenghan.haobaobei I/System.out: result = 2
10-08 17:00:12.817 6650-6728/com.shenghan.haobaobei I/System.out: --------------
10-08 17:00:12.818 6650-6698/com.shenghan.haobaobei I/System.out: along1 --> 2
10-08 17:00:12.820 6650-6698/com.shenghan.haobaobei I/System.out: along2 --> 1
10-08 17:00:12.821 6650-6698/com.shenghan.haobaobei I/System.out: result = 3
10-08 17:00:12.821 6650-6698/com.shenghan.haobaobei I/System.out: --------------
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: along1 --> 2
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: along2 --> 2
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: result = 4
10-08 17:00:13.315 6650-6698/com.shenghan.haobaobei I/System.out: --------------

結(jié)論：

** 1.只有當(dāng)兩個(gè)observable都發(fā)射過第一項(xiàng)數(shù)據(jù)時(shí)，才能進(jìn)行組合發(fā)射**

從log日志來看，observable1打印的第一項(xiàng)數(shù)據(jù)為1，所以可以推斷observable1發(fā)射的0數(shù)據(jù)時(shí)，observable2并沒有發(fā)射過數(shù)據(jù)，所以并沒有進(jìn)行組合。

** 2.當(dāng)observable1與observable2都發(fā)射首個(gè)數(shù)據(jù)后，在此發(fā)射任何數(shù)據(jù)都會(huì)找相應(yīng)的另外一個(gè)observable的最新數(shù)據(jù)進(jìn)行組合發(fā)射**

Join

無論何時(shí)，如果一個(gè)Observable發(fā)射了一個(gè)數(shù)據(jù)項(xiàng)，只要在另一個(gè)Observable發(fā)射的數(shù)據(jù)項(xiàng)定義的時(shí)間窗口內(nèi)，就將兩個(gè)Observable發(fā)射的數(shù)據(jù)合并發(fā)射~

        Observable<Long> observable1 = Observable.interval(2000, 1000, TimeUnit.MILLISECONDS).take(3);
        Observable<Long> observable2 = Observable.interval(2500, 1000, TimeUnit.MILLISECONDS).take(3);

        observable1
                .join(
                        observable2,
                        new Func1<Long, Observable<Long>>() {
                            @Override
                            public Observable<Long> call(Long aLong) {
                                //使Observable延遲500毫秒執(zhí)行
                                System.out.println("observable1 -- >" + aLong);
                                return Observable.just(aLong).delay(500, TimeUnit.MILLISECONDS);
                            }
                        },
                        new Func1<Long, Observable<Long>>() {
                            @Override
                            public Observable<Long> call(Long aLong) {
                                System.out.println("observable2 -- >" + aLong);
                                return Observable.just(aLong).delay(500, TimeUnit.MILLISECONDS);
                            }
                        },
                        new Func2<Long, Long, Long>() {
                            @Override
                            public Long call(Long aLong1, Long aLong2) {
                                System.out.println("aLong1 = " + aLong1);
                                System.out.println("aLong2 = " + aLong2);
                                return aLong1 + aLong2;
                            }
                        })
                .subscribe(new Action1<Long>() {
                    @Override
                    public void call(Long aLong) {
                        System.out.println("result = " + aLong);
                        System.out.println("--------------");
                    }
                });

運(yùn)行結(jié)果：

10-08 16:47:56.988 20959-21011/com.shenghan.haobaobei I/System.out: observable1 -- >0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: observable2 -- >0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: aLong1 = 0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: aLong2 = 0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: result = 0
10-08 16:47:57.488 20959-21041/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: observable1 -- >1
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: aLong1 = 1
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: aLong2 = 0
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: result = 1
10-08 16:47:57.991 20959-21011/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: observable2 -- >1
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: aLong1 = 1
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: aLong2 = 1
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: result = 2
10-08 16:47:58.490 20959-21041/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: observable1 -- >2
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: aLong1 = 2
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: aLong2 = 1
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: result = 3
10-08 16:47:58.990 20959-21011/com.shenghan.haobaobei I/System.out: --------------
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: observable2 -- >2
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: aLong1 = 2
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: aLong2 = 2
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: result = 4
10-08 16:47:59.489 20959-21041/com.shenghan.haobaobei I/System.out: --------------

結(jié)論：

** 1.只有當(dāng)兩個(gè)observable都發(fā)射過第一項(xiàng)數(shù)據(jù)時(shí)，才能進(jìn)行組合發(fā)射**

從log日志來看，observable1發(fā)射數(shù)據(jù)0時(shí)，組合的call方法并沒有被執(zhí)行，而當(dāng)observable2發(fā)射數(shù)據(jù)0后，組合call方法被回調(diào)了。

** 2.當(dāng)observable1與observable2都發(fā)射首個(gè)數(shù)據(jù)后，在此發(fā)射任何數(shù)據(jù)都會(huì)找相應(yīng)的另外一個(gè)observable的最新數(shù)據(jù)進(jìn)行組合發(fā)射**

Merge##

將兩個(gè)Observable發(fā)射的數(shù)據(jù)按照時(shí)間順序進(jìn)行組合，合并成一個(gè)Observable進(jìn)行發(fā)射~

代碼示例：

        Observable<String> observable1 = Observable.interval(0, 500, TimeUnit.MILLISECONDS).take(3).flatMap(new Func1<Long, Observable<String>>() {
            @Override
            public Observable<String> call(Long aLong) {
                return Observable.just("observable1 -- >" + aLong);
            }
        });
        Observable<String> observable2 = Observable.interval(500, 500, TimeUnit.MILLISECONDS).take(3).flatMap(new Func1<Long, Observable<String>>() {
            @Override
            public Observable<String> call(Long aLong) {
                return Observable.just("observable2 -- >" + aLong);
            }
        });

        Observable
                .merge(observable1, observable2)
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String aString) {
                        System.out.println("result = " + aString);
                    }
                });

運(yùn)行結(jié)果：

10-08 17:11:33.696 16661-16737/com.shenghan.haobaobei I/System.out: result = observable1 -- >0
10-08 17:11:34.194 16661-16737/com.shenghan.haobaobei I/System.out: result = observable1 -- >1
10-08 17:11:34.198 16661-16711/com.shenghan.haobaobei I/System.out: result = observable2 -- >0
10-08 17:11:34.695 16661-16737/com.shenghan.haobaobei I/System.out: result = observable1 -- >2
10-08 17:11:34.698 16661-16711/com.shenghan.haobaobei I/System.out: result = observable2 -- >1
10-08 17:11:35.196 16661-16711/com.shenghan.haobaobei I/System.out: result = observable2 -- >2

StartWith##

在源Observable發(fā)射數(shù)據(jù)之前，先發(fā)射一個(gè)指定的數(shù)據(jù)序列或數(shù)據(jù)項(xiàng)~

代碼示例：

        Observable
                .just(4, 5, 6)
                .startWith(1, 2, 3)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("result = " + integer);
                    }
                });

運(yùn)行結(jié)果：

10-08 17:16:43.077 21830-21830/com.shenghan.haobaobei I/System.out: result = 1
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 2
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 3
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 4
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 5
10-08 17:16:43.078 21830-21830/com.shenghan.haobaobei I/System.out: result = 6

switchOnNext##

把一組Observable轉(zhuǎn)換成一個(gè)Observable，如果在同一個(gè)時(shí)間內(nèi)產(chǎn)生兩個(gè)或多個(gè)Observable產(chǎn)生的數(shù)據(jù)，只發(fā)射最后一個(gè)Observable產(chǎn)生的數(shù)據(jù)~

示例代碼：

        Observable<Observable<String>> observable = Observable.interval(2000, 500, TimeUnit.MILLISECONDS).map(new Func1<Long, Observable<String>>() {
            @Override
            public Observable<String> call(Long aLongOutside) {
                //每隔250毫秒產(chǎn)生一組數(shù)據(jù)（0,1,2,3,4)
                return Observable.interval(0, 250, TimeUnit.MILLISECONDS).map(new Func1<Long, String>() {
                    @Override
                    public String call(Long aLongInside) {
                        return "aLongOutside = " + aLongOutside + "| aLongInside = " + aLongInside;
                    }
                }).take(5);
            }
        }).take(2);

        Observable.switchOnNext(observable).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("result = " + s);
            }
        });

運(yùn)行結(jié)果：

10-08 18:20:14.605 2921-2993/com.shenghan.haobaobei I/System.out: result = aLongOutside = 0| aLongInside = 0
10-08 18:20:14.857 2921-2993/com.shenghan.haobaobei I/System.out: result = aLongOutside = 0| aLongInside = 1
10-08 18:20:15.103 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 0
10-08 18:20:15.353 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 1
10-08 18:20:15.604 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 2
10-08 18:20:15.853 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 3
10-08 18:20:16.104 2921-3055/com.shenghan.haobaobei I/System.out: result = aLongOutside = 1| aLongInside = 4

Zip##

使用一個(gè)指定的函數(shù)將多個(gè)Observable發(fā)射的數(shù)據(jù)組合在一起，然后將這個(gè)函數(shù)的結(jié)果作為單項(xiàng)數(shù)據(jù)發(fā)射，嚴(yán)格周期順序進(jìn)行合并，不能單獨(dú)發(fā)射~

示例代碼：

        Observable<String> observable1 = Observable.just("walid - 1", "walid - 2", "walid - 3");
        Observable<String> observable2 = Observable.just("Jordan - 1", "Jordan - 2", "Jordan - 3", "Jordan - 4");
        Observable.zip(observable1, observable2, new Func2<String, String, String>() {
            @Override
            public String call(String s1, String s2) {
                return s1 + " | " + s2;
            }
        }).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                System.out.println("onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                System.err.println("onError");
            }

            @Override
            public void onNext(String value) {
                System.out.println("onNext --> " + value);
            }
        });

運(yùn)行結(jié)果：

10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onNext --> walid - 1 | Jordan - 1
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onNext --> walid - 2 | Jordan - 2
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onNext --> walid - 3 | Jordan - 3
10-08 18:53:31.772 24793-24793/com.shenghan.haobaobei I/System.out: onCompleted

結(jié)論：

** 1.只有每個(gè)observable都依次產(chǎn)品一輪數(shù)據(jù)時(shí)，才會(huì)統(tǒng)一發(fā)射一次，當(dāng)不會(huì)有完整一輪數(shù)據(jù)時(shí)，視為完成**

從log日志來看，observable1產(chǎn)生的數(shù)據(jù)是與observable2一一對(duì)應(yīng)的，也就是只有observable1與observable2同時(shí)產(chǎn)生數(shù)據(jù)時(shí)才會(huì)統(tǒng)一發(fā)射一次~

** 2.當(dāng)不滿足所有observable都有數(shù)據(jù)可產(chǎn)品時(shí)，視為完成狀態(tài)**

從log日志來看，observable2最后一項(xiàng)數(shù)據(jù)“Jordan - 4”并沒有打印，原因是observable1并沒有數(shù)據(jù)可以產(chǎn)生了，所以不滿足發(fā)射條件，視為完成狀態(tài)~

結(jié)語

組合類操作符，就簡(jiǎn)單介紹到這里，希望能夠?qū)ν瑢W(xué)有所幫助，謝謝~

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 本篇文章介主要紹RxJava中操作符是以函數(shù)作為基本單位，與響應(yīng)式編程作為結(jié)合使用的，對(duì)什么是操作、操作符都有哪些...
    嘎啦果安卓獸閱讀 2,991評(píng)論 0贊 10
  • 創(chuàng)建操作 用于創(chuàng)建Observable的操作符Create通過調(diào)用觀察者的方法從頭創(chuàng)建一個(gè)ObservableEm...
    rkua閱讀 1,958評(píng)論 0贊 1
  • 注：只包含標(biāo)準(zhǔn)包中的操作符，用于個(gè)人學(xué)習(xí)及備忘參考博客：http://blog.csdn.net/maplejaw...
    小白要超神閱讀 2,376評(píng)論 2贊 8
  • 作者寄語 很久之前就想寫一個(gè)專題，專寫Android開發(fā)框架，專題的名字叫 XXX 從入門到放棄 ，沉淀了這么久，...
    戴定康閱讀 7,739評(píng)論 13贊 85
  • 作者： maplejaw本篇只解析標(biāo)準(zhǔn)包中的操作符。對(duì)于擴(kuò)展包，由于使用率較低，如有需求，請(qǐng)讀者自行查閱文檔。 創(chuàng)...
    maplejaw_閱讀 46,218評(píng)論 8贊 93

友情鏈接更多精彩內(nèi)容