RxJava/RxAndroid之快速入門2

前言

歡迎繼續收看《我所理解的RxJava--上手其實很簡單（二）》，上周出了第一篇，各位程序猿大大的反應還不錯，生平第一篇博文能獲得大家的肯定，內心肯定是非常開心的，也堅定了我繼續寫下去的信念，總之，先謝謝大家送出的魚丸...哦不，是收藏。好了，題外話不多說，進入這一篇的主題，本文主要給大家補充一下上一篇遺留的Subject知識，沒看過上一篇的同學。俗話說，“書讀百遍，奇異自見”，看多一遍是一遍，多多益善嘛。溫習完的，請回來繼續聽講。

---

關於Subject

關於Subject，官方文檔的解釋是這樣的：Subject可以看成是一個橋梁或者代理，在某些ReactiveX實現中（如RxJava），它同時充當了Observer和Observable的角色。因為它是一個Observer，它可以訂閱一個或多個Observable；又因為它是一個Observable，它可以轉發它收到(Observe)的數據，也可以發射新的數據。從官方解釋中，我提取出三個要點：

1. 它可以充當Observable；
2. 它可以充當Observer；
3. 它是Observable和Observer之間的橋梁；

接下來對這三個要點解釋一下，但在解釋之前，要先介紹一下Subject的種類， Subject是一個抽象類，不能通過new來實例化Subject，所以Subject有四個實現類，分別為AsyncSubject、BehaviorSubject、PublishSubject和ReplaySubject，每個實現類都有特定的“技能”，下面結合代碼來介紹一下它們各自的“技能”。注意，所有的實現類都由create()方法實例化，無需new,所有的實現類調用onCompleted()或onError(),它的Observer將不再接收數據；

---

Subject的分類解析

AsyncSubject
Observer會接收AsyncSubject的`onComplete()之前的最後一個數據，如果因異常而終止，AsyncSubject將不會釋放任何數據，但是會向Observer傳遞一個異常通知。示例代碼如下：

      AsyncSubject asyncSubject = AsyncSubject.create();
      asyncSubject.onNext("asyncSubject1");
      asyncSubject.onNext("asyncSubject2");
      asyncSubject.onNext("asyncSubject3");  
      asyncSubject.onCompleted();
      asyncSubject.subscribe(new Observer() {
          @Override
          public void onCompleted() {

              LogUtil.log("asyncSubject onCompleted");  //輸出 asyncSubject onCompleted
          }

          @Override
          public void onError(Throwable e) {

              LogUtil.log("asyncSubject onError");  //不輸出（異常才會輸出）
          }

          @Override
          public void onNext(String s) {

              LogUtil.log("asyncSubject:"+s);  //輸出asyncSubject:asyncSubject3
          }
      });

以上代碼，Observer只會接收asyncSubject的onCompleted()被調用前的最後一個數據，即“asyncSubject3”，如果不調用onCompleted()，Subscriber將不接收任何數據。

BehaviorSubject
Observer會接收到BehaviorSubject被訂閱之前的最後一個數據，再接收其他發射過來的數據，如果BehaviorSubject被訂閱之前沒有發送任何數據，則會發送一個默認數據。（注意跟AsyncSubject的區別，AsyncSubject要手動調用onCompleted()，且它的Observer會接收到onCompleted()前發送的最後一個數據，之後不會再接收數據，而BehaviorSubject不需手動調用onCompleted()，它的Observer接收的是BehaviorSubject被訂閱前發送的最後一個數據，兩個的分界點不一樣，且之後還會繼續接收數據。）示例代碼如下：

  BehaviorSubject behaviorSubject = BehaviorSubject.create("default");
  behaviorSubject.onNext("behaviorSubject1");
  behaviorSubject.onNext("behaviorSubject2");
      behaviorSubject.subscribe(new Observer() {
          @Override
          public void onCompleted() {

              LogUtil.log("behaviorSubject:complete");
          }

          @Override
          public void onError(Throwable e) {

              LogUtil.log("behaviorSubject:error");
          }

          @Override
          public void onNext(String s) {

              LogUtil.log("behaviorSubject:"+s);
          }
      });

      behaviorSubject.onNext("behaviorSubject3");
      behaviorSubject.onNext("behaviorSubject4");

以上代碼，Observer會接收到behaviorSubject2、behaviorSubject3、behaviorSubject4，如果在behaviorSubject.subscribe()之前不發送behaviorSubject1、behaviorSubject2，則Observer會先接收到default,再接收behaviorSubject3、behaviorSubject4。

PublishSubject

PublishSubject比較容易理解，相對比其他Subject常用，它的Observer只會接收到PublishSubject被訂閱之後發送的數據。示例代碼如下：

  PublishSubject publishSubject = PublishSubject.create();
  publishSubject.onNext("publishSubject1");
  publishSubject.onNext("publishSubject2");
  publishSubject.subscribe(new Observer() {
          @Override
          public void onCompleted() {

          }

          @Override
          public void onError(Throwable e) {

          }

          @Override
          public void onNext(String s) {
              LogUtil.log("publishSubject observer1:"+s);
          }
      });
  publishSubject.onNext("publishSubject3");
  publishSubject.onNext("publishSubject4");

以上代碼，Observer只會接收到"behaviorSubject3"、"behaviorSubject4"。

ReplaySubject
ReplaySubject會發射所有數據給觀察者，無論它們是何時訂閱的。也有其它版本的ReplaySubject，在重放緩存增長到一定大小的時候或過了一段時間後會丟棄舊的數據。示例代碼如下：

ReplaySubjectreplaySubject = ReplaySubject.create(); //創建默認初始緩存容量大小為16的ReplaySubject，當數據條目超過16會重新分配內存空間，使用這種方式，不論ReplaySubject何時被訂閱，Observer都能接收到數據
//replaySubject = ReplaySubject.create(100);//創建指定初始緩存容量大小為100的ReplaySubject
//replaySubject = ReplaySubject.createWithSize(2);//只緩存訂閱前最後發送的2條數據 
//replaySubject=ReplaySubject.createWithTime(1,TimeUnit.SECONDS,Schedulers.computation());  //replaySubject被訂閱前的前1秒內發送的數據才能被接收     
replaySubject.onNext("replaySubject:pre1");
replaySubject.onNext("replaySubject:pre2");
replaySubject.onNext("replaySubject:pre3");
replaySubject.subscribe(new Action1() {
      @Override
      public void call(String s) {
              LogUtil.log("replaySubject:" + s);
      }
  });
replaySubject.onNext("replaySubject:after1");
replaySubject.onNext("replaySubject:after2");

以上代碼，由於情況比較多，注釋也已解釋的相當清楚，就不對輸出結果一一表述了，有疑問的自行copy代碼去測試一下。至此，四種Subject類型已經介紹完畢，但是需要注意，如果你把 Subject 當作一個 Subscriber 使用，不要從多個線程中調用它的onNext方法（包括其它的on系列方法），這可能導致同時（非順序）調用，這會違反Observable協議，給Subject的結果增加了不確定性。要避免此類問題，官方提出了“串行化”，你可以將 Subject 轉換為一個 SerializedSubject ，類似於這樣：

SerializedSubject ser = new SerializedSubject(publishSubject);

---

要點解答

接下來，我們繼續前面提出的問題，為什麼說Subject既可充當Observable，又可充當Observer，是它們兩個之間的橋梁呢？經過前面的例子，也許有些人已經大概理解了，不理解的且聽我細細道來。首先，從理論上講，Subject繼承了Observable，又實現了Observer接口，所以說它既是Observable又是Observer，完全合理。從實際應用上講，Subject也能實現Observable和Observer相同的功能，口說無憑，我們還是通過代碼來證實比較有說服力。

創建Observable並發射數據：

  Observable.create(new Observable.OnSubscribe() {
          @Override
          public void call(Subscriber subscriber) {

              subscriber.onNext("I'm Observable");
              subscriber.onCompleted();
          }
      });

用Subject實現為：

PublishSubject publishSubject = PublishSubject.create();
publishSubject.onNext("as Observable");
publishSubject.onCompleted();

創建Observer訂閱Observable並接收數據：

mObservable.subscribe(new Observer() {
      @Override
       public void onCompleted() {

      }

      @Override
      public void onError(Throwable e) {

      }

      @Override
      public void onNext(String s) {

              //接收數據
      }
});

用Subject實現為：

   publishSubject.subscribe(new Observer() {
          @Override
          public void onCompleted() {

          }

          @Override
          public void onError(Throwable e) {

          }

          @Override
          public void onNext(String s) {

          }
      });

也許有人會問，不是說Subject也可以作為Observer，不能把Subject當作Observer傳入subscribe()中嗎？回答是：當然可以！就象這樣：

PublishSubject publishSubject = PublishSubject.create();
Observable.create(new Observable.OnSubscribe() {
      @Override
      public void call(Subscriber subscriber) {

          subscriber.onNext("as Observer"); 
          subscriber.onCompleted();
      }
}).subscribe(publishSubject);

有沒有發現問題？publishSubject沒有重寫onNext()方法啊，在哪接收的數據？這就是前面說的“橋梁”的問題了，盡管把Subject作為Observer傳入subscribe()，但接收數據還是要通過Observer來接收，借用Subject來連接Observable和Observer，整體代碼如下：

PublishSubject publishSubject = PublishSubject.create();
   Observable.create(new Observable.OnSubscribe() {
          @Override
          public void call(Subscriber subscriber) {

              subscriber.onNext("as Bridge");
              subscriber.onCompleted();
          }
      }).subscribe(publishSubject);

      publishSubject.subscribe(new Observer() {
          @Override
          public void onCompleted() {

          }

          @Override
          public void onError(Throwable e) {

          }

          @Override
          public void onNext(String s) {

              LogUtil.log("subject:"+s); //接收到 as Bridge
          }
      });

沒錯，這很橋梁！

---

總結

關於Subject，到此就介紹完了。也許你會跟我一樣困惑，為什麼又要多個Subject出來，除了有幾個特定功能之外，其他所有的一切，Observable和Observer也都有，而且寫法上也沒有原來的簡便。確實如此，對於幾個特定功能，我也還想不到有什麼應用場景，至少我還沒發現有什麼場景必須得用Subject來實現不可，那麼問題又來了，我為什麼要花這麼大篇幅來介紹Subject，理由有三。其一，既然官方推出Subject，必有其道理，還沒遇到不代表以後不會遇到，更不能代表你不會遇到這樣的應用場景；其二，“一千個讀者有一千個哈姆雷特”，我所看到的並不是全部，也許你會發掘出更有意思的東西可不是？其三，我可不想當你看完我所有關於RxJava的文章，自信已上手RxJava，當有人跟你提起Subject的時候，你一臉茫然不知道Subject是什麼東西，豈不哀哉？所以呢，介紹一下Subject還是很有意義的，最起碼學了比沒學好，“養兵千日用兵一時”，知識不嫌多，突然哪天就用上了呢。對於Subject的理解，有異議的歡迎底下評論，一起交流進步。下一篇文章，進入RxJava操作符的使用講解。