RxJava操作符系列二（下）
接上文
輸出日志信息
- call:2 ConcatMap RxNewThreadScheduler-5
- onNext: ConcatMap 101 ConcatMap
- call:2 ConcatMap RxNewThreadScheduler-6
- onNext: ConcatMap 102 ConcatMap
- call:2 ConcatMap RxNewThreadScheduler-7
- onNext: ConcatMap 103 ConcatMap
- onCompleted: ConcatMap
通過該操作符和flatMap輸出的日志信息，很容易看出flatMap并沒有保證數(shù)據(jù)源的順序性，但是ConcatMap操作符保證了數(shù)據(jù)源的順序性。在應(yīng)用中，如果你對數(shù)據(jù)的順序性有要求的話，就需要使用ConcatMap。若沒有要求，二者皆可使用。
SwitchMap
當(dāng)原始Observable發(fā)射一個(gè)新的數(shù)據(jù)(Observable)時(shí)，它將取消訂閱并停止監(jiān)視產(chǎn)生執(zhí)之前那個(gè)數(shù)據(jù)的Observable，只監(jiān)視當(dāng)前這一個(gè).
- Integer[] integers = {1, 2, 3};
- Observable.from(integers).switchMap(new Func1>() {
- @Override
- public Observable call(Integer integer) {
- Log.e(TAG, "call: SwitchMap" + Thread.currentThread().getName());
- //如果不通過subscribeOn(Schedulers.newThread())在在子線程模擬并發(fā)操作，所有數(shù)據(jù)源依然會(huì)全部輸出,也就是并發(fā)操作此操作符才有作用
- //若在此通過Thread。sleep（）設(shè)置等待時(shí)間，則輸出信息會(huì)不一樣。相當(dāng)于模擬并發(fā)程度
- return Observable.just((integer + 100) + "SwitchMap").subscribeOn(Schedulers.newThread());
- }
- }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted: SwitchMap");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError: SwitchMap");
- }
- @Override
- public void onNext(String s) {
- Log.e(TAG, "onNext: SwitchMap "+s);
- }
- });
輸出日志信息
- call: SwitchMapmain
- call: SwitchMapmain
- call: SwitchMapmain
- onNext: SwitchMap 106SwitchMap
- onCompleted: SwitchMap
當(dāng)數(shù)據(jù)源較多時(shí)，并不一定是只輸出***一項(xiàng)數(shù)據(jù)，有可能輸出幾項(xiàng)數(shù)據(jù)，也可能是全部。
GroupBy
看到這個(gè)詞你就應(yīng)該想到了這個(gè)操作符的作用，就是你理解的含義，他將數(shù)據(jù)源按照你的約定進(jìn)行分組。我們通過groupBy實(shí)行將1到10的數(shù)據(jù)進(jìn)行就劃分，代碼如下
- Observable.range(1, 10).groupBy(new Func1() {
- @Override
- public Boolean call(Integer integer) {
- return integer % 2 == 0;
- }
- }).subscribe(new Subscriber>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted:1 ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError:1 ");
- }
- @Override
- public void onNext(GroupedObservable booleanIntegerGroupedObservable) {
- booleanIntegerGroupedObservable.toList().subscribe(new Subscriber>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted:2 " );
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError:2 ");
- }
- @Override
- public void onNext(List integers) {
- Log.e(TAG, "onNext:2 "+integers);
- }
- });
- }
- });
輸出日志信息
- onNext:2 [1, 3, 5, 7, 9]
- onCompleted:2
- onNext:2 [2, 4, 6, 8, 10]
- onCompleted:2
- onCompleted:1
在上面代碼中booleanIntegerGroupedObservable變量有一個(gè)getKey()方法,該方法返回的是分組的key，他的值就是groupBy方法call回調(diào)所用函數(shù)的值，在上面也就是integer % 2 == 0的值，及true和false。有幾個(gè)分組也是有此值決定的。
Scan
操作符對原始Observable發(fā)射的***項(xiàng)數(shù)據(jù)應(yīng)用一個(gè)函數(shù)，然后將那個(gè)函數(shù)的結(jié)果作為自己的***項(xiàng)數(shù)據(jù)發(fā)射。它將函數(shù)的結(jié)果同第二項(xiàng)數(shù)據(jù)一起填充給這個(gè)函數(shù)來產(chǎn)生它自己的第二項(xiàng)數(shù)據(jù)。它持續(xù)進(jìn)行這個(gè)過程來產(chǎn)生剩余的數(shù)據(jù)序列。
例如計(jì)算1+2+3+4的和
- Observable.range(1,4).scan(new Func2() {
- @Override
- public Integer call(Integer integer, Integer integer2) {
- Log.e(TAG, "call: integer:"+integer+" integer2 "+integer2);
- return integer+integer2;
- }
- }).subscribe(new Subscriber() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted: ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError: " );
- }
- @Override
- public void onNext(Integer integer) {
- Log.e(TAG, "onNext: "+integer );
- }
- });
輸出日志信息
- onNext: 1
- call: integer:1 integer2 2
- onNext: 3
- call: integer:3 integer2 3
- onNext: 6
- call: integer:6 integer2 4
- onNext: 10
- onCompleted:
對于scan有一個(gè)重載方法，可以設(shè)置一個(gè)初始值，如上面代碼，初始值設(shè)置為10，只需將scan加個(gè)參數(shù)scan(10，new Func2)。
Buffer
操作符將一個(gè)Observable變換為另一個(gè)，原來的Observable正常發(fā)射數(shù)據(jù)，變換產(chǎn)生的Observable發(fā)射這些數(shù)據(jù)的緩存集合，如果原來的Observable發(fā)射了一個(gè)onError通知，Buffer會(huì)立即傳遞這個(gè)通知，而不是首先發(fā)射緩存的數(shù)據(jù)，即使在這之前緩存中包含了原始Observable發(fā)射的數(shù)據(jù)。
示例代碼
- Observable.range(10, 6).buffer(2).subscribe(new Subscriber>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted: ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError: ");
- }
- @Override
- public void onNext(List integers) {
- Log.e(TAG, "onNext: " + integers);
- }
- });
輸出日志信息
- onNext: [10, 11]
- onNext: [12, 13]
- onNext: [14, 15]
- onCompleted:
上面一次性訂閱兩個(gè)數(shù)據(jù)，如果設(shè)置參數(shù)為6，就一次性訂閱。buffer的另一重載方法buffer(count, skip)從原始Observable的***項(xiàng)數(shù)據(jù)開始創(chuàng)建新的緩存(長度count)，此后每當(dāng)收到skip項(xiàng)數(shù)據(jù)，用count項(xiàng)數(shù)據(jù)填充緩存：開頭的一項(xiàng)和后續(xù)的count-1項(xiàng)，它以列表(List)的形式發(fā)射緩存，取決于count和skip的值，這些緩存可能會(huì)有重疊部分(比如skip count時(shí))。具體執(zhí)行結(jié)果，你可以設(shè)置不同的skip和count觀察輸出日志，查看執(zhí)行結(jié)果及流程。
Window
Window和Buffer類似，但不是發(fā)射來自原始Observable的數(shù)據(jù)包，它發(fā)射的是Observables，這些Observables中的每一個(gè)都發(fā)射原始Observable數(shù)據(jù)的一個(gè)子集，***發(fā)射一個(gè)onCompleted通知。
- Observable.range(10, 6).window(2).subscribe(new Subscriber>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted1: ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError1: ");
- }
- @Override
- public void onNext(Observable integerObservable) {
- Log.e(TAG, "onNext1: ");
- tv1.append("\n");
- integerObservable.subscribe(new Subscriber() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted2: ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError2: ");
- }
- @Override
- public void onNext(Integer integer) {
- Log.e(TAG, "onNext2: "+integer);
- }
- });
- }
- });
輸出日志信息
- onNext2: 10
- onNext2: 11
- onCompleted2:
- onNext2: 12
- onNext2: 13
- onCompleted2:
- onNext2: 14
- onNext2: 15
- onCompleted2:
- onCompleted1:
window和buffer一樣也有不同的重載方法。這兩個(gè)操作符相對其他操作符不太容易理解，可以去RxJava GitHub理解，里面有圖示解析。當(dāng)然***的理解方式就是通過更改變量的值，去觀察輸出的日志信息。
好了，這篇文章就介紹到這里。若文中有錯(cuò)誤的地方，歡迎指正。謝謝。
