RxJava系列5(組合操作符)

阿新 • • 發佈：2018-04-04

nds lec ber letter ava esc err 結果 src

這一章我們接著介紹組合操作符，這類operators可以同時處理多個Observable來創建我們所需要的Observable。組合操作符主要包含： Merge StartWith Concat Zip CombineLatest SwitchOnNext Join等等。

Merge

merge(Observable, Observable)將兩個Observable發射的事件序列組合並成一個事件序列，就像是一個Observable發射的一樣。你可以簡單的將它理解為兩個Obsrvable合並成了一個Observable，合並後的數據是無序的。

merge(Observable, Observable)

我們看下面的例子，一共有兩個Observable：一個用來發送字母，另一個用來發送數字；現在我們需要兩連個Observable發射的數據合並。

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);

Observable.merge(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }

            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString()+" ");
            }
        });

程序輸出：

A 0 B C 1 D E 2 F 3 G H 4

merge(Observable[])將多個Observable發射的事件序列組合並成一個事件序列，就像是一個Observable發射的一樣。

merge(Observable[])

StartWith

startWith(T)用於在源Observable發射的數據前插入數據。使用startWith(Iterable<T>)我們還可以在源Observable發射的數據前插入Iterable。官方示意圖：

startWith(T) startWith(Iterable<T>)

startWith(Observable<T>)

用於在源Observable發射的數據前插入另一個Observable發射的數據（這些數據會被插入到
源Observable發射數據的前面）。官方示意圖：

startWith(Observable<T>)

Concat

concat(Observable<? extends T>, Observable<? extends T>) concat(Observable<？ extends Observable<T>>)用於將多個obserbavle發射的的數據進行合並發射，concat嚴格按照順序發射數據，前一個Observable沒發射玩是不會發射後一個Observable的數據的。它和merge、startWitch和相似，不同之處在於：

merge:合並後發射的數據是無序的；
startWitch:只能在源Observable發射的數據前插入數據。

concat(Observable<? extends T>, Observable<? extends T>)、concat(Observable<？ extends Observable<T>>)

這裏我們將前面Merge操作符的例子拿過來，並將操作符換成Concat，然後我們看看執行結果：

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);

Observable.concat(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }

            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString() + " ");
            }
        });

程序輸出：

A B C D E F G H 0 1 2 3 4

Zip

zip(Observable, Observable, Func2)用來合並兩個Observable發射的數據項，根據Func2函數生成一個新的值並發射出去。當其中一個Observable發送數據結束或者出現異常後，另一個Observable也將停在發射數據。

zip(Observable, Observable, Func2)

和前面的例子一樣，我們將操作符換成了zip:

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(120, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);

Observable<Long> numberSequence = Observable.interval(200, TimeUnit.MILLISECONDS).take(5);

Observable.zip(letterSequence, numberSequence, new Func2<String, Long, String>() {
    @Override
    public String call(String letter, Long number) {
        return letter + number;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("Error:" + e.getMessage());
    }

    @Override
    public void onNext(String result) {
        System.out.print(result + " ");
    }
});

程序輸出：

A0 B1 C2 D3 E4

CombineLatest

comnineLatest(Observable, Observable, Func2)用於將兩個Observale最近發射的數據已經Func2函數的規則進展組合。下面是官方提供的原理圖：

comnineLatest(Observable, Observable, Func2)

下面這張圖應該更容易理解：

comnineLatest(Observable, Observable, Func2)

List<String> communityNames = DataSimulator.getCommunityNames();
List<Location> locations = DataSimulator.getLocations();

Observable<String> communityNameSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return communityNames.get(position.intValue());
            }
        }).take(communityNames.size());
Observable<Location> locationSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, Location>() {
            @Override
            public Location call(Long position) {
                return locations.get(position.intValue());
            }
        }).take(locations.size());

Observable.combineLatest(
        communityNameSequence,
        locationSequence,
        new Func2<String, Location, String>() {
            @Override
            public String call(String communityName, Location location) {
                return "小區名:" + communityName + ", 經緯度:" + location.toString();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }
        });

程序輸出：

小區名:竹園新村, 經緯度:(21.827, 23.323)
小區名:康橋半島, 經緯度:(21.827, 23.323)
小區名:康橋半島, 經緯度:(11.923, 16.309)
小區名:中糧·海景壹號, 經緯度:(11.923, 16.309)
小區名:中糧·海景壹號, 經緯度:(22.273, 53.623)
小區名:浦江名苑, 經緯度:(22.273, 53.623)
小區名:南輝小區, 經緯度:(22.273, 53.623)

SwitchOnNext

switchOnNext(Observable<? extends Observable<? extends T>>用來將一個發射多個小Observable的源Observable轉化為一個Observable，然後發射這多個小Observable所發射的數據。如果一個小的Observable正在發射數據的時候，源Observable又發射出一個新的小Observable，則前一個Observable發射的數據會被拋棄，直接發射新
的小Observable所發射的數據。

結合下面的原理圖大家應該很容易理解，我們可以看到下圖中的黃色圓圈就被丟棄了。

switchOnNext(Observable<? extends Observable<? extends T>>)

Join

join(Observable, Func1, Func1, Func2)我們先介紹下join操作符的4個參數：

Observable：源Observable需要組合的Observable,這裏我們姑且稱之為目標Observable；
Func1：接收從源Observable發射來的數據，並返回一個Observable，這個Observable的聲明周期決定了源Obsrvable發射出來的數據的有效期；
Func1：接收目標Observable發射來的數據，並返回一個Observable，這個Observable的聲明周期決定了目標Obsrvable發射出來的數據的有效期；
Func2：接收從源Observable和目標Observable發射出來的數據，並將這兩個數據組合後返回。

所以Join操作符的語法結構大致是這樣的：onservableA.join(observableB, 控制observableA發射數據有效期的函數，控制observableB發射數據有效期的函數，兩個observable發射數據的合並規則)

join操作符的效果類似於排列組合，把第一個數據源A作為基座窗口，他根據自己的節奏不斷發射數據元素，第二個數據源B，每發射一個數據，我們都把它和第一個數據源A中已經發射的數據進行一對一匹配；舉例來說，如果某一時刻B發射了一個數據“B”,此時A已經發射了0，1，2，3共四個數據，那麽我們的合並操作就會把“B”依次與0,1,2,3配對，得到四組數據： [0, B] [1, B] [2, B] [3, B]

再看看下面的圖是不是好理解了呢？！

join(Observable, Func1, Func1, Func2)

讀懂了上面的文字，我們再來寫段代碼加深理解。

final List<House> houses = DataSimulator.getHouses();//模擬的房源數據，用於測試

//用來每秒從houses總取出一套房源並發射出去
Observable<House> houseSequence =
        Observable.interval(1, TimeUnit.SECONDS)
                .map(new Func1<Long, House>() {
                    @Override
                    public House call(Long position) {
                        return houses.get(position.intValue());
                    }
                }).take(houses.size());//這裏的take是為了防止houses.get(position.intValue())數組越界

//用來實現每秒發送一個新的Long型數據
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);

houseSequence.join(tictoc,
        new Func1<House, Observable<Long>>() {
            @Override
            public Observable<Long> call(House house) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        },
        new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                return Observable.timer(0, TimeUnit.SECONDS);
            }
        },
        new Func2<House, Long, String>() {
            @Override
            public String call(House house, Long aLong) {
                return aLong + "-->" + house.getDesc();
            }
        }
).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("Error:"+e.getMessage());
    }

    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});

程序輸出：

0-->中糧海景壹號新出大平層！總價4500W起
1-->中糧海景壹號新出大平層！總價4500W起
1-->滿五唯一，黃金地段
2-->中糧海景壹號新出大平層！總價4500W起
2-->滿五唯一，黃金地段
2-->一樓自帶小花園
3-->一樓自帶小花園
3-->毗鄰湯臣一品
4-->毗鄰湯臣一品
4-->頂級住宅，給您總統般尊貴體驗
5-->頂級住宅，給您總統般尊貴體驗
5-->頂層戶型，兩室一廳
6-->頂層戶型，兩室一廳
6-->南北通透，豪華五房
7-->南北通透，豪華五房

通過轉換操作符、過濾操作符、組合操作符三個篇幅將RxJava主要的操作符也介紹的七七八八了。更多操作符的介紹建議大家去查閱官方文檔，並自己動手實踐一下。這一系列的文章也會持續更新，歡迎大家保持關註！:)

Demo源碼地址：https://github.com/BaronZ88/HelloRxJava

鏈接：https://www.jianshu.com/p/546fe44a6e22

RxJava系列5(組合操作符)

nds lec ber letter ava esc err 結果 src 這一章我們接著介紹組合操作符，這類operators可以同時處理多個Observable來創建我們所需要的Observable。組合操作符主要包含： Merge StartWith Concat

RxJava系列5(組合操作符)

Merge

StartWith

Concat

Zip

CombineLatest

SwitchOnNext

Join

RxJava系列5(組合操作符)

RxJava系列4(過濾操作符)

RxJava系列3(轉換操作符)

RxJava 系列之變換操作符flatmap(2)

Android RxJava操作符詳解系列：變換操作符

Android函式響應式程式設計——必學的RxJava組合操作符startWith、merge、concat、zip、combineLastest

RxJava 變換、組合、合併操作符

Swift新手教程系列5-函數+selector在swift中的使用方法

ACdream區域賽指導賽之手速賽系列(5) 題解

C++遊戲系列5：不止有一件武器

[js高手之路]深入淺出webpack教程系列5-插件使用之html-webpack-plugin配置(中)

Python學習系列-----第二章操作符與表達式

Django-website 程序案例系列-5 模態對話框實現提交數據

【系列5】使用Dockerfile創建帶weblogic的Centos Docker鏡像

智能合約語言 Solidity 教程系列5 - 數組介紹

Office 365實現單點登錄系列(5)—配置單點登錄

Azure手把手系列5：Azure帳戶和訂閱

批處理系列(5) - 根據文件名自動創建目錄存放文件

scrapy框架系列 (5) Spider類

RxJava系列6(從微觀角度解讀RxJava源碼)

RxJava系列5(組合操作符)

Merge

StartWith

Concat

Zip

CombineLatest

SwitchOnNext

Join

相關推薦