一、ReactiveX簡單介紹

Rx是一個使用可觀察資料流進行非同步程式設計的程式設計介面，ReactiveX結合了觀察者模式、迭代器模式和函數語言程式設計的精華。Rx提供了一系列的操作符，你可以使用它們來過濾(filter)、選擇(select)、變換(transform)、結合(combine)和組合(compose)多個Observable，這些操作符讓執行和複合變得非常高效。

在ReactiveX中，一個觀察者(Observer)訂閱一個可觀察物件(Observable)。觀察者對Observable發射的資料或資料序列作出響應。這種模式可以極大地簡化併發操作，因為它建立了一個處於待命狀態的觀察者哨兵，在未來某個時刻響應Observable的通知，不需要阻塞等待Observable發射資料。

上圖取自ReactiveX官方文件，上面一排圖示代表被觀察物件產生的事件，橫向的箭頭代表時間線，有6個事件依次發射，經過中間的轉換和處理得到了下面的圖示，即處理結果，有些事件發射、處理和接收都成功，而有些事件因為各種原因導致失敗，這些情況都會在相應的回撥方法中呈現。

Subscribe方法用於將觀察者連線到Observable，你的觀察者需要實現以下方法的一個子集：

• onNext(T item)

  Observable呼叫這個方法發射資料，方法的引數就是Observable發射的資料，這個方法可能會被呼叫多次，取決於你的實現。

• onError(Exception ex)

  當Observable遇到錯誤或者無法返回期望的資料時會呼叫這個方法，這個呼叫會終止Observable，後續不會再呼叫onNext和onCompleted，onError方法的引數是丟擲的異常。

• onComplete

  正常終止，如果沒有遇到錯誤，Observable在最後一次呼叫onNext之後呼叫此方法。

根據Observable協議的定義，onNext可能會被呼叫零次或者很多次，最後會有一次onCompleted或onError呼叫（不會同時），傳遞資料給onNext通常被稱作發射，onCompleted和onError被稱作通知。

二、RxJava與RxAndroid的配合使用

RxJava是 ReactiveX 在JVM上的一個實現，ReactiveX使用Observable序列組合非同步和基於事件的程式。RxJava提供了5種排程器，分別是：

• .io()

  這個排程器時用於I/O操作。它基於根據需要，增長或縮減來自適應的執行緒池。由於它專用於I/O操作，所以並不是RxJava的預設方法；正確的使用它是由開發者決定的。重點需要注意的是執行緒池是無限制的，大量的I/O排程操作將建立許多個執行緒並佔用記憶體。

• .computation()

  這個是計算工作預設的排程器，它與I/O操作無關。它也是許多RxJava方法的預設排程器：buffer(), debounce(), delay() , interval(), sample(), skip()。

• .immediate()

  這個排程器允許你立即在當前執行緒執行你指定的工作。它是timeout(),timeInterval(),以及timestamp()方法預設的排程器。

• .newThread()

  這個排程器為指定任務啟動一個新的執行緒。

• .trampoline()

  當我們想在當前執行緒執行一個任務時，並不是立即，我們可以用.trampoline()將它入隊。這個排程器將會處理它的佇列並且按序執行佇列中每一個任務。它是repeat()和retry()方法預設的排程器。

RxAndroid模組包含RxJava的Android特定的繫結程式碼。它給RxJava添加了一些類，用於幫助在Android應用中編寫響應式(reactive)的元件。它提供了一個可以在給定的Android Handler上排程Observable的排程器 Scheduler，特別是在UI主執行緒上 AndroidSchedulers.mainThread()。

上面提到了幾種執行緒排程器，可以讓開發者在不同的執行緒執行不同的事件，那麼如何指定和切換執行緒呢？RxJava提供了兩個方法subscribeOn和observeOn，前者可以指定Observable事件產生和變換處理的執行緒，後者可以指定訂閱者所在的執行緒，假如事件是一個耗時任務，完全可以通過subscribeOn指定為計算執行緒或者子執行緒，然後再通過observeOn切換Android主執行緒，即可在訂閱者回調方法中操作UI檢視。

看一個具體的例子吧。

Observable.just("")
  .subscribeOn(Schedulers.newThread())
  .map(s -> {
    for (PerformanceEntity.Components components : performanceEntity.components) {
        PerformanceSubmitEntity.Data submitData = new PerformanceSubmitEntity.Data();
        submitData.componentId = components.id;
        for (PerformanceEntity.Data data : performanceEntity.data) {
            if (data.componentId == components.id) {
                transferData(submitData, data);
                break;
            }
        }
        submitDataList.add(submitData);
    }
    return null;
  })
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(o -> {
    initPerformanceView();
  });

在渲染檢視之前需要處理大量的資料，那麼就可以將Observable的執行緒切換至子執行緒，等負責的業務邏輯完成後再切換至UI執行緒，然後在subscribe中完成檢視的渲染。這樣即可以保證UI執行緒不會大量複雜的計算，也可以不用Handler、AsycTask等複雜的操作，程式碼清晰度也相對較高。

三、RxJava與Retrofit的結合

在Android開發中，網路請求往往是最耗時的，也是情況最複雜的，如果直接在UI執行緒進行網路請求，編譯時就會報錯，及時編譯報錯，執行時系統也會報ANR錯誤，所以網路請求必須在子執行緒完成。而網路請求結果一般又需要操作UI檢視，所以返回結果的回撥有必須在主執行緒，那麼上面提到的方法就可以完美解決這個問題。

網路請求的返回結果一般都是json格式，如果返回值為原始的字串，那麼就需要呼叫放每次都要進行json至JavaBean的轉換，有的時候我們只關心返回結果中的一部分資料，那麼就還需要在返回結果中再次對資料進行篩選，還有情況是我們只關心滿足一定條件的資料，那麼這些需求如果用RxJava來實現就變得異常簡單。有關Retrofit詳細用法，以及如何將返回結果切換成RxJava處理模式，可以閱讀Retrofit用法詳解這篇文章，下面直接看示例：

 showProgressBar();
 messageService.messageList(msgType, PAGE_INDEX, PAGE_SIZE)
            .compose(new DefaultTransformer<>(getActivity()))
            .map(messageListResponse -> messageListResponse.data)
            .flatMap(messageListEntity -> Observable.from(messageListEntity))
            .filter(messageEntity.id > 10)
            .subscribe(new Subscriber<MessageEntity>() {
                @Override
                public void onCompleted() {
                    dismissProgressBar();
                    refreshLayout.setRefreshing(false);
                }
                @Override
                public void onError(Throwable e) {
                    dismissProgressBar();
                    refreshLayout.setRefreshing(false);
                }
                @Override
                public void onNext(MessageEntity messageEntity) {
                    updateView(messageEntity);
                }
            });

• DefaultTransformer主要實現執行緒的切換以及錯誤訊息的統一處理，一般來說服務端介面返回的資料格式都比較固定，會有code、message、data，其中code只是正確或者錯誤型別，message返回正確或者錯誤日誌，data則是業務資料的存放地，那麼就可以在DefaultTransformer中統一判斷code值，如果是請求出錯，那麼直接將異常拋給onError回撥方法。
• map中實現的功能就是上文提到的呼叫方只關心data中的資料，而不關心code和message，那麼就可以通過map方法直接將data中的資料傳遞給訂閱者。
• 如果我們想將返回值List逐條處理，一般的做法就是直接for迴圈，這裡用到了flatMap，就是將messageListEntity通過Observable的from方法重新拆分成更細粒度的MessageEntity，而訂閱者得到也就是MessageEntity。
• filter方法就是通過布林表示式篩選出符合條件的資料，上述例子中就是將id值大於10的MessageEntity篩選出來。

有關Observable的compose、map、from、flatMap、filter等方法可以參考ReactiveX官方教程，中文環境下可以參考給Android開發者的RxJava詳解

歡迎大家關注我的微信公眾號藍田大營，下面是微信公眾號二維碼：