如果你做過Android（和Java）的開發，很有可能已經聽說過RxJava了。RxJava是由Netflix開發的響應式擴展（Reactive Extensions）的Java實現。引用MSDN上對它的定義，Reactive Extensions是這樣一個第三方庫：它結合了可觀察集合和LINQ式查詢以達到異步和基于事件的編程效果。Netflix將這個庫托管到了Github上，支持Java6以上的版本并且使它可以用于Android App開發。

本篇是介紹RxJava和Android的系列文章的***篇，將會介紹如何在Android中使用RxJava observables（基于Square的Retrofit組件）創建REST API客戶端。

[[107490]]

我們從添加所需的庫文件開始。如果你用Maven的話，只需將下面的dependencies（依賴庫）加到pom.xml中即可：

<dependency> 
    <groupId>com.squareup.retrofit</groupId> 
    <artifactId>retrofit</artifactId> 
    <version>1.2.2</version> 
</dependency> 
<dependency> 
    <groupId>com.netflix.rxjava</groupId> 
    <artifactId>rxjava-android</artifactId> 
    <version>0.14.6</version> 
</dependency> 

在本文中，我們將用氣象地圖開放平臺（OpenWeatherMap） API作為演示示例。 OpenWeatherMap是一個免費的天氣數據API，非常易于配置和使用，調用時只需傳入位置信息（城市名或者是地理坐標）作為參數即可，具體效果請參見這個示例。它默認傳輸的是JSON格式的數據（但也可以配置為XML或HTML格式）。精度和溫度單位也是可以配置的，更多詳情請看這里。

通常要實現調用一個API需要如下這幾個步驟（每個步驟都有一堆公式化代碼）：

1. 創建所需的模型類（必要時，添加上注解）。
2. 實現請求—回應管理的網絡層代碼，并帶錯誤處理。
3. 用后臺線程實現請求調用（一般是用異步任務的形式實現），用一個回調函數（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組件來實現的話將大大減少代碼量。只需要創建一個接口類（用注釋來描述整個請求），然后用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部分！我們這里示例將用它來實現異步的請求調用。但這并不是RxJava所有的功能，以下對RxJava的介紹引用自Netflix的Github 知識庫：

RxJava 是一個在Java虛擬機上實現的響應式擴展庫：提供了基于observable序列實現的異步調用及基于事件編程。

它擴展了觀察者模式，支持數據、事件序列并允許你合并序列，無需關心底層的線程處理、同步、線程安全、并發數據結構和非阻塞I/O處理。

它支持Java5及更高版本，并支持其他一些基于JVM的語言，如Groovy、Clojure和Scala。

我們假設你已經對RxJava有一些了解。如果沒有的話，強烈建議先看看這兩篇 文章和Netflix在Github Wiki上的前幾頁。

在***的這個示例中，我們將實現一個API 管理器負責生成observable對象，并完成多并發調用（每個調用都請求同一個地址，但參數不同）。

首先我們需要將前面創建的接口類，換為這個類：

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

還想看點其他資料不？看看這篇文章吧，它講的是RxJava如何進行錯誤處理。