RxJava入門指南及其在Android開發中的使用示例

RxJava的GitHub主頁，部署部分就沒什麼好說的了~
https://github.com/ReactiveX/RxJava

基礎
RxJava最核心的兩個東西是Observables（被觀察者，事件源）和Subscribers（觀察者）。Observables發出一系列事件，Subscribers處理這些事件。這裡的事件可以是任何你感興趣的東西（觸摸事件，web接口調用返回的數據。。。）

一個Observable可以發出零個或者多個事件，知道結束或者出錯。每發出一個事件，就會調用它的Subscriber的onNext方法，最後調用Subscriber.onNext()或者Subscriber.onError()結束。

Rxjava的看起來很想設計模式中的觀察者模式，但是有一點明顯不同，那就是如果一個Observerble沒有任何的的Subscriber，那麼這個Observable是不會發出任何事件的。

Hello World
創建一個Observable對象很簡單，直接調用Observable.create即可

Observable<String> myObservable = Observable.create( 
  new Observable.OnSubscribe<String>() { 
    @Override 
    public void call(Subscriber<? super String> sub) { 
      sub.onNext("Hello, world!"); 
      sub.onCompleted(); 
    } 
  } 
); 

這裡定義的Observable對象僅僅發出一個Hello World字符串，然後就結束了。接著我們創建一個Subscriber來處理Observable對象發出的字符串。

Subscriber<String> mySubscriber = new Subscriber<String>() { 
  @Override 
  public void onNext(String s) { System.out.println(s); } 
 
  @Override 
  public void onCompleted() { } 
 
  @Override 
  public void onError(Throwable e) { } 
}; 

這裡subscriber僅僅就是打印observable發出的字符串。通過subscribe函數就可以將我們定義的myObservable對象和mySubscriber對象關聯起來，這樣就完成了subscriber對observable的訂閱。

myObservable.subscribe(mySubscriber); 

一旦mySubscriber訂閱了myObservable，myObservable就是調用mySubscriber對象的onNext和onComplete方法，mySubscriber就會打印出Hello World！

更簡潔的代碼
是不是覺得僅僅為了打印一個hello world要寫這麼多代碼太啰嗦？我這裡主要是為了展示RxJava背後的原理而采用了這種比較啰嗦的寫法，RxJava其實提供了很多便捷的函數來幫助我們減少代碼。

首先來看看如何簡化Observable對象的創建過程。RxJava內置了很多簡化創建Observable對象的函數，比如Observable.just就是用來創建只發出一個事件就結束的Observable對象，上面創建Observable對象的代碼可以簡化為一行

Observable<String> myObservable = Observable.just("Hello, world!"); 
接下來看看如何簡化Subscriber，上面的例子中，我們其實並不關心OnComplete和OnError，我們只需要在onNext的時候做一些處理，這時候就可以使用Action1類。

Action1<String> onNextAction = new Action1<String>() { 
  @Override 
  public void call(String s) { 
    System.out.println(s); 
  } 
}; 

subscribe方法有一個重載版本，接受三個Action1類型的參數，分別對應OnNext，OnComplete， OnError函數。 

myObservable.subscribe(onNextAction, onErrorAction, onCompleteAction); 

這裡我們並不關心onError和onComplete，所以只需要第一個參數就可以

 
myObservable.subscribe(onNextAction); 
// Outputs "Hello, world!" 

上面的代碼最終可以寫成這樣

Observable.just("Hello, world!") 
  .subscribe(new Action1<String>() { 
    @Override 
    public void call(String s) { 
       System.out.println(s); 
    } 
  }); 

使用java8的lambda可以使代碼更簡潔
 

Observable.just("Hello, world!") 
  .subscribe(s -> System.out.println(s)); 

Android開發中，強烈推薦使用retrolambda這個gradle插件，這樣你就可以在你的代碼中使用lambda了。

變換
讓我們做一些更有趣的事情吧！
比如我想在hello world中加上我的簽名，你可能會想到去修改Observable對象：

Observable.just("Hello, world! -Dan") 
  .subscribe(s -> System.out.println(s)); 

如果你能夠改變Observable對象，這當然是可以的，但是如果你不能修改Observable對象呢？比如Observable對象是第三方庫提供的？比如我的Observable對象被多個Subscriber訂閱，但是我只想在對某個訂閱者做修改呢？
那麼在Subscriber中對事件進行修改怎麼樣呢？比如下面的代碼：

Observable.just("Hello, world!") 
  .subscribe(s -> System.out.println(s + " -Dan")); 

這種方式仍然不能讓人滿意，因為我希望我的Subscribers越輕量越好，因為我有可能會在mainThread中運行subscriber。另外，根據響應式函數編程的概念，Subscribers更應該做的事情是“響應”，響應Observable發出的事件，而不是去修改。如果我能在某些中間步驟中對“Hello World！”進行變換是不是很酷？

實例

我們將用氣象地圖開放平台（OpenWeatherMap） API作為演示示例。 OpenWeatherMap（http://api.openweathermap.org/）是一個免費的天氣數據API，非常易於配置和使用，調用時只需傳入位置信息（城市名或者是地理坐標）作為參數即可
通常要實現調用一個API需要如下這幾個步驟（每個步驟都有一堆公式化代碼）：

創建所需的模型類（必要時，添加上注解）。
實現請求—回應管理的網絡層代碼，並帶錯誤處理。
用後台線程實現請求調用（一般是用異步任務的形式實現），用一個回調函數（Callback Function）來實現在UI線程上呈現回應信息。
創建模型類

第一步我們可以依靠一些類似jsonschema2pojo的JSON-POJO生成工具（半）自動化完成。OpenWeather API的模型類如下：

public class WeatherData {
 
  public Coordinates coord;
  public Local sys;
  public List<Weather> weathers;
  public String base;
  public Main main;
  public Wind wind;
  public Rain rain;
  public Cloud clouds;
  public long id;
  public long dt;
  public String name;
  public int cod;
 
  public static class Coordinates {
    public double lat;
    public double lon;
  }
 
  public static class Local {
    public String country;
    public long sunrise;
    public long sunset;
  }
 
  public static class Weather {
    public int id;
    public String main;
    public String description;
    public String icon;
  }
 
  public static class Main {
    public double temp;
    public double pressure;
    public double humidity;
    public double temp_min;
    public double temp_max;
    public double sea_level;
    public double grnd_level;
  }
 
  public static class Wind {
    public double speed;
    public double deg;
  }
 
  public static class Rain {
    public int threehourforecast;
  }
 
  public static class Cloud {
    public int all;
  }
 
}

用Retrofit實現網絡調用

第二步中網絡調用的實現通常我們需要寫一大堆公式化的代碼，但如果用Square公司的Retrofit組件（http://square.github.io/retrofit/）來實現的話將大大減少代碼量。只需要創建一個接口類（用注釋來描述整個請求），然後用RestAdapter.Builder來創建客戶端就行了。Retrofit也可以用來完成JSON的序列化與反序列化。

private interface ApiManagerService {
  @GET("/weather")
  WeatherData getWeather(@Query("q") String place, @Query("units") String units);
}

上面的示例中我們可以看到，方法前的注釋是由一個HTTP方法（我們這裡用的是GET，當然你也可以按需要用Retrofit實現POST、PUT、DELETE和HEAD方法）和一個相對路徑（基本路徑是由RestAdapter.Builder提供的）。@Query注釋用於組裝請求參數，我們這有兩個參數，一個是place（代表位置），另一個是units計量單位。

我們來看一個具體的調用示例（實際代碼中應該把這個調用放到一個非UI線程裡）。這段代碼還是比較容易理解的：

//...
final RestAdapter restAdapter = new RestAdapter.Builder()
  .setServer("http://api.openweathermap.org/data/2.5")
  .build();
 
final ApiManagerService apiManager = restAdapter.create(ApiManagerService.class);
final WeatherData weatherData = apiManager.getWeather("Budapest,hu", "metric");
//...

怎麼樣，很簡單吧，你只需要很少的代碼就實現了整個調用過程，這就是Retrofit的威力

用RxJava實現響應式編程

現在我們就進入第三步了：RxJava部分！我們這裡示例將用它來實現異步的請求調用。
首先我們需要將前面創建的接口類，換為這個類：

public class ApiManager {
 
  private interface ApiManagerService {
    @GET("/weather")
    WeatherData getWeather(@Query("q") String place, @Query("units") String units);
  }
 
  private static final RestAdapter restAdapter = new RestAdapter.Builder()
    .setServer("http://api.openweathermap.org/data/2.5")
    .build();
  private static final ApiManagerService apiManager = restAdapter.create(ApiManagerService.class);
 
  public static Observable<WeatherData> getWeatherData(final String city) {
    return Observable.create(new Observable.OnSubscribeFunc<WeatherData>() {
      @Override
      public Subscription onSubscribe(Observer<? super WeatherData> observer) {
        try {
          observer.onNext(apiManager.getWeather(city, "metric"));
          observer.onCompleted();
        } catch (Exception e) {
          observer.onError(e);
        }
 
        return Subscriptions.empty();
      }
    }).subscribeOn(Schedulers.threadPoolForIO());
  }
 
}

我們先來看下getWeatherData()這個方法，它調用了Observable.create()方法並向方法傳入一個Observable.OnSubscribeFunc的實現，以此得到一個Observable對象並返回。並且一旦Observable對象被訂閱（subscribed）後就會開始工作。Observable每次處理的結果都會當作參數傳給onNext()方法。因為我們這裡只是想實現網絡請求的並發調用，所以只需要讓每個Observable對象中調用一次請求即可。代碼最後調用onComplete()方法。這裡的subscribeOn()方法很重要，它決定了程序將選用哪種線程。這裡調用的是Schedulers.threadPoolForIO()，此線程用於優化IO和網絡性能相關的工作。

最後一步是要實現這個API調用。下面的代碼實現了並發網絡請求，每個請求都使用不同的調用參數異步調用同一個url：

Observable.from(cities)
      .mapMany(new Func1<String, Observable<WeatherData>>() {
        @Override
        public Observable<WeatherData> call(String s) {
          return ApiManager.getWeatherData(s);
        }
      })
      .subscribeOn(Schedulers.threadPoolForIO())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action1<WeatherData>() {
        @Override
        public void call(WeatherData weatherData) {
          // do your work
        }
      });

Observable.from()方法將城市名稱數組轉化為一個observable對象，將數組裡的字符串提供給不同的線程。然後mapMany()方法將會把前者提供的每一個字符串都轉化為observable對象（譯注：新對象包含的是weatherData對象數據）。這裡的轉化通過調用ApiManager.getWeatherData()完成。

這裡還是注冊在I/O線程池上。在Android系統上，如果需要把結果展示在UI上，就必須把數據發布給UI線程處理。因為我們知道，在Android上只有最原始的那個創建界面的線程才可以操作界面。這裡只需要用observeOn()方法調用AndroidSchedulers.mainThread()即可。subscribe()方法的調用將觸發observable對象，我們可以在這裡處理observable對象發出的結果。

這個示例展示了RxJava強大的功能。如果沒有Rx，我們需要創建N個線程去調用請求，然後通過異步方式把處理結果交給UI線程。使用Rx只需編寫很少的代碼就完成工作，使用它強大的功能創建、合並、過濾和轉化observable對象。

RxJava可以在開發安卓App時，作為一個強大的處理並發的工具使用。雖然要熟悉它還是需要一些時間，但是磨刀不誤砍柴工，一旦掌握了它，將給你帶來很大幫助。響應式擴展庫是個很好的想法，我們把它用於安卓程序的開發，已經用了好幾個禮拜了（在不久的將來，我們產品的異步任務處理將完全基於它完成）。越是了解它，你就越會愛上它。