如果把整個(gè)事件看作是工廠的流水線，Observable就是原料。Observer就是我們的產(chǎn)品經(jīng)理，這個(gè)產(chǎn)品是怎么交到我們產(chǎn)品經(jīng)理手上的呢？ 中間很重要的就是工人，也就是操作符。它負(fù)責(zé)在Observable發(fā)出的事件和Observable的響應(yīng)之間做一些處理。

Map

用來把一個(gè)事件轉(zhuǎn)換為另一個(gè)事件，

Observable.just("RxJava")     
   .map(new Func1<String, Integer>() {
        @Override
        public Integer call(String s) {
            return 2016;                    //將String轉(zhuǎn)為Integer            
        }      
    }).map(new Func1<Integer, String>() {            
        @Override
        public String call(Integer integer) {
            return integer.toString();      //再通過第二個(gè)map轉(zhuǎn)成String
        }      
    }).subscribe(new Action1<String>() {
        @Override            
        public void call(String s) {
            System.out.println(s);
        }   
    });

From

接收一個(gè)集合作為輸入，然后每次輸出一個(gè)元素給subscriber.

//from
List<String> s = Arrays.asList("Java", "Android", "Ruby", "Ios", "Swift");
  Observable.from(s)
            .subscribe(new Action1<String>() {
              @Override
              public void call(String s) {
                  System.out.println(s);
                }
          });

FlatMap

Observable.flatMap()接收一個(gè)Observable的輸出作為輸入，同時(shí)輸出另外一個(gè)Observable。

先加一個(gè)函數(shù)

static Observable<List<String>>query(){
       List<String> s = Arrays.asList("Java", "Android", "Ruby", "Ios", "Swift");
       return Observable.just(s);  
}

我們打印所有query到的語言:

// 注意這里的參數(shù)是 query所返回的Observable的輸出,并且返會(huì)一個(gè)Observable<String>
query().flatMap(new Func1<List<String>, Observable<String>>() {
       @Override
       public Observable<String> call(List<String> strings) { //結(jié)合from處理
             return Observable.from(strings);
           }
       }).subscribe(new Action1<String>() {
       @Override public void call(String s) {
             System.out.println("_flatMap:"+s);
           }
      });

假設(shè)，我們要在輸出的String上加個(gè)前綴，在不修改Obserble的基礎(chǔ)上怎么修改.

增加一個(gè)方法，用來增加前綴

    static Observable<String>addPre(String lan){
           return Observable.just("addPre_"+lan);
      }

代碼就可以這么寫

query().flatMap(new Func1<List<String>, Observable<String>>() {
     @Override public Observable<String> call(List<String> strings) {
             return Observable.from(strings);
       }
    }).flatMap(new Func1<String, Observable<String>>() {
     @Override public Observable<String> call(String s) { //我們在這里調(diào)用`addPre`方法,就行處理         
          return addPre(s);
       }
    }).subscribe(new Action1<String>() {
   @Override public void call(String s) { 
      System.out.println(s);
       }
  });

更多操作符

filter 過濾,把不符合條件的過濾掉,留下符合條件的
take 指定最多輸出的數(shù)量
doOnNext 允許我們在每次輸出一個(gè)元素之前做一些額外的事情(其實(shí)就是在onNext
里調(diào)用的)

就用一個(gè)例子來演示一下吧:

query().flatMap(new Func1<List<String>, Observable<? extends String>>() {
     @Override
     public Observable<? extends String> call(List<String> strings) {
      return Observable.from(strings);
     }
 }).filter(new Func1<String, Boolean>() {
  @Override
public Boolean call(String s) {
    return s.contains("a");    //只發(fā)射包含“a”的事件
}
 }).take(3).doOnNext(new Action1<String>() {
    @Override
    public void call(String s) {

    }
}).subscribe(new Action1<String>() {
    @Override
    public void call(String s1) {
      System.out.println("_flatMap:" + s1);
    }
});

操作符決策樹
幾種主要的需求
直接創(chuàng)建一個(gè)Observable（創(chuàng)建操作）
組合多個(gè)Observable（組合操作）
對Observable發(fā)射的數(shù)據(jù)執(zhí)行變換操作（變換操作）
從Observable發(fā)射的數(shù)據(jù)中取特定的值（過濾操作）
轉(zhuǎn)發(fā)Observable的部分值（條件/布爾/過濾操作）
對Observable發(fā)射的數(shù)據(jù)序列求值（算術(shù)/聚合操作）
創(chuàng)建操作
用于創(chuàng)建Observable的操作符

• Create — 通過調(diào)用觀察者的方法從頭創(chuàng)建一個(gè)Observable
• Defer — 在觀察者訂閱之前不創(chuàng)建這個(gè)Observable，為每一個(gè)觀察者創(chuàng)建一個(gè)新的Observable
• Empty/Never/Throw — 創(chuàng)建行為受限的特殊Observable
• From — 將其它的對象或數(shù)據(jù)結(jié)構(gòu)轉(zhuǎn)換為Observable
• Interval — 創(chuàng)建一個(gè)定時(shí)發(fā)射整數(shù)序列的Observable
• Just — 將對象或者對象集合轉(zhuǎn)換為一個(gè)會(huì)發(fā)射這些對象的Observable
• Range — 創(chuàng)建發(fā)射指定范圍的整數(shù)序列的Observable
• Repeat — 創(chuàng)建重復(fù)發(fā)射特定的數(shù)據(jù)或數(shù)據(jù)序列的Observable
• Start — 創(chuàng)建發(fā)射一個(gè)函數(shù)的返回值的Observable
• Timer — 創(chuàng)建在一個(gè)指定的延遲之后發(fā)射單個(gè)數(shù)據(jù)的Observable

變換操作
這些操作符可用于對Observable發(fā)射的數(shù)據(jù)進(jìn)行變換，詳細(xì)解釋可以看每個(gè)操作符的文檔

• Buffer — 緩存，可以簡單的理解為緩存，它定期從Observable收集數(shù)據(jù)到一個(gè)集合，然后把這些數(shù)據(jù)集合打包發(fā)射，而不是一次發(fā)射一個(gè)
• FlatMap — 扁平映射，將Observable發(fā)射的數(shù)據(jù)變換為Observables集合，然后將這些Observable發(fā)射的數(shù)據(jù)平坦化的放進(jìn)一個(gè)單獨(dú)的Observable，可以認(rèn)為是一個(gè)將嵌套的數(shù)據(jù)結(jié)構(gòu)展開的過程。
• GroupBy — 分組，將原來的Observable分拆為Observable集合，將原始Observable發(fā)射的數(shù)據(jù)按Key分組，每一個(gè)Observable發(fā)射一組不同的數(shù)據(jù)
• Map — 映射，通過對序列的每一項(xiàng)都應(yīng)用一個(gè)函數(shù)變換Observable發(fā)射的數(shù)據(jù)，實(shí)質(zhì)是對序列中的每一項(xiàng)執(zhí)行一個(gè)函數(shù)，函數(shù)的參數(shù)就是這個(gè)數(shù)據(jù)項(xiàng)
• Scan — 掃描，對Observable發(fā)射的每一項(xiàng)數(shù)據(jù)應(yīng)用一個(gè)函數(shù)，然后按順序依次發(fā)射這些值
• Window — 窗口，定期將來自O(shè)bservable的數(shù)據(jù)分拆成一些Observable窗口，然后發(fā)射這些窗口，而不是每次發(fā)射一項(xiàng)。類似于Buffer，但Buffer發(fā)射的是數(shù)據(jù)，Window發(fā)射的是Observable，每一個(gè)Observable發(fā)射原始Observable的數(shù)據(jù)的一個(gè)子集

過濾操作
這些操作符用于從Observable發(fā)射的數(shù)據(jù)中進(jìn)行選擇

• Debounce — 只有在空閑了一段時(shí)間后才發(fā)射數(shù)據(jù)，通俗的說，就是如果一段時(shí)間沒有操作，就執(zhí)行一次操作
• Distinct — 去重，過濾掉重復(fù)數(shù)據(jù)項(xiàng)
• ElementAt — 取值，取特定位置的數(shù)據(jù)項(xiàng)
• Filter — 過濾，過濾掉沒有通過謂詞測試的數(shù)據(jù)項(xiàng)，只發(fā)射通過測試的
• First — 首項(xiàng)，只發(fā)射滿足條件的第一條數(shù)據(jù)
• IgnoreElements — 忽略所有的數(shù)據(jù)，只保留終止通知(onError或onCompleted)
• Last — 末項(xiàng)，只發(fā)射最后一條數(shù)據(jù)
• Sample — 取樣，定期發(fā)射最新的數(shù)據(jù)，等于是數(shù)據(jù)抽樣，有的實(shí)現(xiàn)里叫ThrottleFirst
• Skip — 跳過前面的若干項(xiàng)數(shù)據(jù)
• SkipLast — 跳過后面的若干項(xiàng)數(shù)據(jù)
• Take — 只保留前面的若干項(xiàng)數(shù)據(jù)
• TakeLast — 只保留后面的若干項(xiàng)數(shù)據(jù)

組合操作
組合操作符用于將多個(gè)Observable組合成一個(gè)單一的Observable

• And/Then/When — 通過模式(And條件)和計(jì)劃(Then次序)組合兩個(gè)或多個(gè)Observable發(fā)射的數(shù)據(jù)集
• CombineLatest — 當(dāng)兩個(gè)Observables中的任何一個(gè)發(fā)射了一個(gè)數(shù)據(jù)時(shí)，通過一個(gè)指定的函數(shù)組合每個(gè)>* >* Observable發(fā)射的最新數(shù)據(jù)（一共兩個(gè)數(shù)據(jù)），然后發(fā)射這個(gè)函數(shù)的結(jié)果
• 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ā)射
• Merge — 將兩個(gè)Observable發(fā)射的數(shù)據(jù)組合并成一個(gè)
• StartWith — 在發(fā)射原來的Observable的數(shù)據(jù)序列之前，先發(fā)射一個(gè)指定的數(shù)據(jù)序列或數(shù)據(jù)項(xiàng)
• Switch — 將一個(gè)發(fā)射Observable序列的Observable轉(zhuǎn)換為這樣一個(gè)Observable：它逐個(gè)發(fā)射那些>* Observable最近發(fā)射的數(shù)據(jù)
• Zip — 打包，使用一個(gè)指定的函數(shù)將多個(gè)Observable發(fā)射的數(shù)據(jù)組合在一起，然后將這個(gè)函數(shù)的結(jié)果作為單項(xiàng)數(shù)據(jù)發(fā)射

錯(cuò)誤處理
這些操作符用于從錯(cuò)誤通知中恢復(fù)

• Catch — 捕獲，繼續(xù)序列操作，將錯(cuò)誤替換為正常的數(shù)據(jù)，從onError通知中恢復(fù)
• Retry — 重試，如果Observable發(fā)射了一個(gè)錯(cuò)誤通知，重新訂閱它，期待它正常終止

輔助操作
一組用于處理Observable的操作符

• Delay — 延遲一段時(shí)間發(fā)射結(jié)果數(shù)據(jù)
• Do — 注冊一個(gè)動(dòng)作占用一些Observable的生命周期事件，相當(dāng)于Mock某個(gè)操作
• Materialize/Dematerialize — 將發(fā)射的數(shù)據(jù)和通知都當(dāng)做數(shù)據(jù)發(fā)射，或者反過來
• ObserveOn — 指定觀察者觀察Observable的調(diào)度程序（工作線程）
• Serialize — 強(qiáng)制Observable按次序發(fā)射數(shù)據(jù)并且功能是有效的
• Subscribe — 收到Observable發(fā)射的數(shù)據(jù)和通知后執(zhí)行的操作
• SubscribeOn — 指定Observable應(yīng)該在哪個(gè)調(diào)度程序上執(zhí)行
• TimeInterval — 將一個(gè)Observable轉(zhuǎn)換為發(fā)射兩個(gè)數(shù)據(jù)之間所耗費(fèi)時(shí)間的Observable
• Timeout — 添加超時(shí)機(jī)制，如果過了指定的一段時(shí)間沒有發(fā)射數(shù)據(jù)，就發(fā)射一個(gè)錯(cuò)誤通知
• Timestamp — 給Observable發(fā)射的每個(gè)數(shù)據(jù)項(xiàng)添加一個(gè)時(shí)間戳
• Using — 創(chuàng)建一個(gè)只在Observable的生命周期內(nèi)存在的一次性資源

條件和布爾操作
這些操作符可用于單個(gè)或多個(gè)數(shù)據(jù)項(xiàng)，也可用于Observable

• All — 判斷Observable發(fā)射的所有的數(shù)據(jù)項(xiàng)是否都滿足某個(gè)條件
• Amb — 給定多個(gè)Observable，只讓第一個(gè)發(fā)射數(shù)據(jù)的Observable發(fā)射全部數(shù)據(jù)
• Contains — 判斷Observable是否會(huì)發(fā)射一個(gè)指定的數(shù)據(jù)項(xiàng)
• DefaultIfEmpty — 發(fā)射來自原始Observable的數(shù)據(jù)，如果原始Observable沒有發(fā)射數(shù)據(jù)，就發(fā)射一個(gè)默認(rèn)數(shù)據(jù)
• SequenceEqual — 判斷兩個(gè)Observable是否按相同的數(shù)據(jù)序列
• SkipUntil — 丟棄原始Observable發(fā)射的數(shù)據(jù)，直到第二個(gè)Observable發(fā)射了一個(gè)數(shù)據(jù)，然后發(fā)射原始Observable的剩余數(shù)據(jù)
• SkipWhile — 丟棄原始Observable發(fā)射的數(shù)據(jù)，直到一個(gè)特定的條件為假，然后發(fā)射原始Observable剩余的數(shù)據(jù)
• TakeUntil — 發(fā)射來自原始Observable的數(shù)據(jù)，直到第二個(gè)Observable發(fā)射了一個(gè)數(shù)據(jù)或一個(gè)通知
• TakeWhile — 發(fā)射原始Observable的數(shù)據(jù)，直到一個(gè)特定的條件為真，然后跳過剩余的數(shù)據(jù)

算術(shù)和聚合操作
這些操作符可用于整個(gè)數(shù)據(jù)序列

• Average — 計(jì)算Observable發(fā)射的數(shù)據(jù)序列的平均值，然后發(fā)射這個(gè)結(jié)果
• Concat — 不交錯(cuò)的連接多個(gè)Observable的數(shù)據(jù)
• Count — 計(jì)算Observable發(fā)射的數(shù)據(jù)個(gè)數(shù)，然后發(fā)射這個(gè)結(jié)果
• Max — 計(jì)算并發(fā)射數(shù)據(jù)序列的最大值
• Min — 計(jì)算并發(fā)射數(shù)據(jù)序列的最小值
• Reduce — 按順序?qū)?shù)據(jù)序列的每一個(gè)應(yīng)用某個(gè)函數(shù)，然后返回這個(gè)值
• Sum — 計(jì)算并發(fā)射數(shù)據(jù)序列的和

連接操作
一些有精確可控的訂閱行為的特殊Observable

• Connect — 指示一個(gè)可連接的Observable開始發(fā)射數(shù)據(jù)給訂閱者
• Publish — 將一個(gè)普通的Observable轉(zhuǎn)換為可連接的
• RefCount — 使一個(gè)可連接的Observable表現(xiàn)得像一個(gè)普通的Observable
• Replay — 確保所有的觀察者收到同樣的數(shù)據(jù)序列，即使他們在Observable開始發(fā)射數(shù)據(jù)之后才訂閱

轉(zhuǎn)換操作

• To — 將Observable轉(zhuǎn)換為其它的對象或數(shù)據(jù)結(jié)構(gòu)

Blocking 阻塞Observable的操作符