[[180037]]

接上文

輸出日志信息

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

當原始Observable發(fā)射一個新的數(shù)據(jù)(Observable)時，它將取消訂閱并停止監(jiān)視產(chǎn)生執(zhí)之前那個數(shù)據(jù)的Observable，只監(jiān)視當前這一個.

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ù)源依然會全部輸出,也就是并發(fā)操作此操作符才有作用 
                //若在此通過Thread。sleep（）設(shè)置等待時間，則輸出信息會不一樣。相當于模擬并發(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  

當數(shù)據(jù)源較多時，并不一定是只輸出***一項數(shù)據(jù)，有可能輸出幾項數(shù)據(jù)，也可能是全部。

GroupBy

看到這個詞你就應(yīng)該想到了這個操作符的作用，就是你理解的含義，他將數(shù)據(jù)源按照你的約定進行分組。我們通過groupBy實行將1到10的數(shù)據(jù)進行就劃分，代碼如下 

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變量有一個getKey()方法,該方法返回的是分組的key，他的值就是groupBy方法call回調(diào)所用函數(shù)的值，在上面也就是integer % 2 == 0的值，及true和false。有幾個分組也是有此值決定的。

Scan

操作符對原始Observable發(fā)射的***項數(shù)據(jù)應(yīng)用一個函數(shù)，然后將那個函數(shù)的結(jié)果作為自己的***項數(shù)據(jù)發(fā)射。它將函數(shù)的結(jié)果同第二項數(shù)據(jù)一起填充給這個函數(shù)來產(chǎn)生它自己的第二項數(shù)據(jù)。它持續(xù)進行這個過程來產(chǎn)生剩余的數(shù)據(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有一個重載方法，可以設(shè)置一個初始值，如上面代碼，初始值設(shè)置為10，只需將scan加個參數(shù)scan(10，new Func2)。

Buffer

操作符將一個Observable變換為另一個，原來的Observable正常發(fā)射數(shù)據(jù)，變換產(chǎn)生的Observable發(fā)射這些數(shù)據(jù)的緩存集合，如果原來的Observable發(fā)射了一個onError通知，Buffer會立即傳遞這個通知，而不是首先發(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:  

上面一次性訂閱兩個數(shù)據(jù)，如果設(shè)置參數(shù)為6，就一次性訂閱。buffer的另一重載方法buffer(count, skip)從原始Observable的***項數(shù)據(jù)開始創(chuàng)建新的緩存(長度count)，此后每當收到skip項數(shù)據(jù)，用count項數(shù)據(jù)填充緩存：開頭的一項和后續(xù)的count-1項，它以列表(List)的形式發(fā)射緩存，取決于count和skip的值，這些緩存可能會有重疊部分(比如skip count時)。具體執(zhí)行結(jié)果，你可以設(shè)置不同的skip和count觀察輸出日志，查看執(zhí)行結(jié)果及流程。

Window

Window和Buffer類似，但不是發(fā)射來自原始Observable的數(shù)據(jù)包，它發(fā)射的是Observables，這些Observables中的每一個都發(fā)射原始Observable數(shù)據(jù)的一個子集，***發(fā)射一個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一樣也有不同的重載方法。這兩個操作符相對其他操作符不太容易理解，可以去RxJava GitHub理解，里面有圖示解析。當然***的理解方式就是通過更改變量的值，去觀察輸出的日志信息。

好了，這篇文章就介紹到這里。若文中有錯誤的地方，歡迎指正。謝謝。