RxJava操作符（二）Transforming Observables

RxJava操作符（二）Transforming Observables

在上一篇文章中，我們了解了如何創建Observable，僅僅創建一個Observable可能無法滿足一些復雜的場景，所以我們很可能需要將創建的Observable安裝某種規則轉化一下來發射數據。在這篇文章裡我們來了解一下如何來轉化Observable

一、Buffer
顧名思義，Buffer操作符所要做的事情就是將數據安裝規定的大小做一下緩存，然後將緩存的數據作為一個集合發射出去。如下圖所示，第一張示例圖中我們指定buffer的大小為3，收集到3個數據後就發射出去，第二張圖中我們加入了一個skip參數用來指定每次發射一個集合需要跳過幾個數據，圖中如何指定count為2，skip為3，就會每3個數據發射一個包含兩個數據的集合，如果count==skip的話，我們就會發現其等效於第一種情況了。


buffer不僅僅可以通過數量規則來緩存，還可以通過時間等規則來緩存，如規定3秒鐘緩存發射一次等，見下面代碼，我們創建了兩個Observable，並使用buffer對其進行轉化，第一個通過數量來緩存，第二個通過時間來緩存。

private Observable<List<Integer>> bufferObserver() {
return Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).buffer(2, 3);
}

private Observable<List<Long>> bufferTimeObserver() {
return Observable.interval(1, TimeUnit.SECONDS).buffer(3, TimeUnit.SECONDS).observeOn(AndroidSchedulers.mainThread());
}

對其進行訂閱

mLButton.setText("buffer");
mLButton.setOnClickListener(e -> bufferObserver().subscribe(i -> log("buffer:" + i)));
mRButton.setText("bufferTime");
mRButton.setOnClickListener(e -> bufferTimeObserver().subscribe(i -> log("bufferTime:" + i)));

運行結果如下，可以看到第一個Observable會每隔3個數字發射出前兩個數字；第二個Observable會每隔3秒鐘輸出2~4個數字。


二、FlatMap
FlatMap是一個用處多多的操作符，可以將要數據根據你想要的規則進行轉化後再發射出去。其原理就是將這個Observable轉化為多個以原Observable發射的數據作為源數據的Observable，然後再將這多個Observable發射的數據整合發射出來，需要注意的是最後的順序可能會交錯地發射出來，如果對順序有嚴格的要求的話可以使用concatmap操作符。FlatMapIterable和FlatMap基相同，不同之處為其轉化的多個Observable是使用Iterable作為源數據的。


下面我們分別使用FlatMap和FlatMapIterable創建並轉化兩個Observable。

private Observable<String> flatMapObserver() {
return Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).flatMap(integer -> Observable.just("flat map:" + integer));
}

private Observable<? extends Integer> flatMapIterableObserver() {
return Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
.flatMapIterable(
integer -> {
ArrayList<Integer> s = new ArrayList<>();
for (int i = 0; i < integer; i++) {
s.add(integer);
}
return s;
}
);
}

分別對其進行訂閱

mLButton.setText("flatMap");
mLButton.setOnClickListener(e -> flatMapObserver().subscribe(i -> log(i)));
mRButton.setText("flatMapIterable");
mRButton.setOnClickListener(e -> flatMapIterableObserver().subscribe(i -> log("flatMapIterable:" + i)));

運行後的結果如下所示，第一個操作符將發射的數據都加上了個flat map的字符串前綴，第二個將數據做了擴展，會輸出n個n數字。



三、GroupBy
GroupBy操作符將原始Observable發射的數據按照key來拆分成一些小的Observable，然後這些小的Observable分別發射其所包含的的數據，類似於sql裡面的groupBy。
在使用中，我們需要提供一個生成key的規則，所有key相同的數據會包含在同一個小的Observable種。另外我們還可以提供一個函數來對這些數據進行轉化，有點類似於集成了flatMap。

下面創建兩個經過groupBy轉化的Observable對象，第一個按照奇數偶數分組，第二個分組後將數字加上一個字符串前綴

mLButton.setText("groupBy");
mLButton.setOnClickListener(e -> groupByObserver().subscribe(new Subscriber<GroupedObservable<Integer, Integer>>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(GroupedObservable<Integer, Integer> groupedObservable) {
groupedObservable.count().subscribe(integer -> log("key" + groupedObservable.getKey() + " contains:" + integer + " numbers"));
}
}));
mRButton.setText("groupByKeyValue");
mRButton.setOnClickListener(e -> groupByKeyValueObserver().subscribe(new Subscriber<GroupedObservable<Integer, String>>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(GroupedObservable<Integer, String> integerIntegerGroupedObservable) {
if (integerIntegerGroupedObservable.getKey() == 0) {
integerIntegerGroupedObservable.subscribe(integer -> log(integer));
}
}
}));
}

運行結果如下，我們拿到想要的結果。



四、Map、Cast
Map操作符的功能類似於FlatMap，不同之處在於它對數據的轉化是直接進行的，而FlatMap需要通過一些中間的Observables來進行。

Cast將Observable發射的數據強制轉化為另外一種類型，屬於Map的一種具體的實現

下面我們創建兩個經過map和cast轉化的Observable對象

private Observable<Integer> mapObserver() {
return Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).map(integer -> integer * 10);
}

private Observable<Dog> castObserver() {
return Observable.just(getAnimal())
.cast(Dog.class);
}

Animal getAnimal() {
return new Dog();
}

class Animal {
protected String name = "Animal";

Animal() {
log("create " + name);
}

String getName() {
return name;
}
}

class Dog extends Animal {
Dog() {
name = getClass().getSimpleName();
log("create " + name);
}

}

對其進行注冊

mLButton.setText("Map");
mLButton.setOnClickListener(e -> mapObserver().subscribe(i -> log("Map:" + i)));
mRButton.setText("Cast");
mRButton.setOnClickListener(e -> castObserver().subscribe(i -> log("Cast:" + i.getName())));

運行後得到結果如下。可以看到，map操作符將數據都乘以10後再發射出來，cast操作符將Animal類型的對象強制轉化為Dog類型的對象。另外我們還可以驗證一下一個知識點，有繼承的情況下創建對象會首先調用父類的構造方法哦。


五、Scan
Scan操作符對一個序列的數據應用一個函數，並將這個函數的結果發射出去作為下個數據應用這個函數時候的第一個參數使用，有點類似於遞歸操作

下面我們通過一個存放10個2的list創建一個Observable對象並使用scan對其進行轉化，轉化的函數就是計算的結果乘以下一個數。

private Observable<Integer> scanObserver() {
return Observable.from(list).scan((x, y) -> x * y).observeOn(AndroidSchedulers.mainThread());
}

對其進行訂閱

mLButton.setText("scan");
mLButton.setOnClickListener(e -> scanObserver().subscribe(i -> log("scan:" + i)));

得到結果如下，可以看到，我們輸出了2的n次方。


六、Window
Window操作符類似於我們前面講過的buffer，不同之處在於window發射的是一些小的Observable對象，由這些小的Observable對象來發射內部包含的數據。同buffer一樣，window不僅可以通過數目來分組還可以通過時間等規則來分組

下面我們創建兩個Observable對象分別使用window的數目和時間規則來進行分組。

private Observable<Observable<Integer>> windowCountObserver() {
return Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).window(3);
}

private Observable<Observable<Long>> wondowTimeObserver() {
return Observable.interval(1000, TimeUnit.MILLISECONDS)
.window(3000, TimeUnit.MILLISECONDS)
.observeOn(AndroidSchedulers.mainThread());
}

分別對其訂閱

mLButton.setText("window");
mLButton.setOnClickListener(e -> windowCountObserver().subscribe(i -> {
log(i);
i.subscribe((j -> log("window:" + j)));
}));
mRButton.setText("Time");
mRButton.setOnClickListener(e -> wondowTimeObserver().subscribe(i -> {
log(i);
i.observeOn(AndroidSchedulers.mainThread()).subscribe((j -> log("wondowTime:" + j)));
}));

運行結果如下，可以看到第一個Observable對象沒次發射出一個包含3個數據的小Observable，第二個Observable對象每隔3秒鐘發射出一個包含2~4個數據的Observable對象


Transforming操作符是Rxjava強大之處的重要體現，要靈活使用Rxjava掌握Transforming操作符是必不可少的。

本文的demo程序見github:https://github.com/Chaoba/RxJavaDemo