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

基礎(chǔ)
RxJava最核心的兩個(gè)東西是Observables（被觀察者，事件源）和Subscribers（觀察者）。Observables發(fā)出一系列事件，Subscribers處理這些事件。這里的事件可以是任何你感興趣的東西（觸摸事件，web接口調(diào)用返回的數(shù)據(jù)。。。）

一個(gè)Observable可以發(fā)出零個(gè)或者多個(gè)事件，知道結(jié)束或者出錯(cuò)。每發(fā)出一個(gè)事件，就會(huì)調(diào)用它的Subscriber的onNext方法，最后調(diào)用Subscriber.onNext()或者Subscriber.onError()結(jié)束。

Rxjava的看起來很想設(shè)計(jì)模式中的觀察者模式，但是有一點(diǎn)明顯不同，那就是如果一個(gè)Observerble沒有任何的的Subscriber，那么這個(gè)Observable是不會(huì)發(fā)出任何事件的。

Hello World
創(chuàng)建一個(gè)Observable對象很簡單，直接調(diào)用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對象僅僅發(fā)出一個(gè)Hello World字符串，然后就結(jié)束了。接著我們創(chuàng)建一個(gè)Subscriber來處理Observable對象發(fā)出的字符串。

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發(fā)出的字符串。通過subscribe函數(shù)就可以將我們定義的myObservable對象和mySubscriber對象關(guān)聯(lián)起來，這樣就完成了subscriber對observable的訂閱。

myObservable.subscribe(mySubscriber); 

一旦mySubscriber訂閱了myObservable，myObservable就是調(diào)用mySubscriber對象的onNext和onComplete方法，mySubscriber就會(huì)打印出Hello World！

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

首先來看看如何簡化Observable對象的創(chuàng)建過程。RxJava內(nèi)置了很多簡化創(chuàng)建Observable對象的函數(shù)，比如Observable.just就是用來創(chuàng)建只發(fā)出一個(gè)事件就結(jié)束的Observable對象，上面創(chuàng)建Observable對象的代碼可以簡化為一行

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

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

subscribe方法有一個(gè)重載版本，接受三個(gè)Action1類型的參數(shù)，分別對應(yīng)OnNext，OnComplete， OnError函數(shù)。 

myObservable.subscribe(onNextAction, onErrorAction, onCompleteAction); 

這里我們并不關(guān)心onError和onComplete，所以只需要第一個(gè)參數(shù)就可以

 
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開發(fā)中，強(qiáng)烈推薦使用retrolambda這個(gè)gradle插件，這樣你就可以在你的代碼中使用lambda了。

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

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

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

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

這種方式仍然不能讓人滿意，因?yàn)槲蚁Ｍ业腟ubscribers越輕量越好，因?yàn)槲矣锌赡軙?huì)在mainThread中運(yùn)行subscriber。另外，根據(jù)響應(yīng)式函數(shù)編程的概念，Subscribers更應(yīng)該做的事情是“響應(yīng)”，響應(yīng)Observable發(fā)出的事件，而不是去修改。如果我能在某些中間步驟中對“Hello World！”進(jìn)行變換是不是很酷？

實(shí)例

我們將用氣象地圖開放平臺(tái)（OpenWeatherMap） API作為演示示例。 OpenWeatherMap（http://api.openweathermap.org/）是一個(gè)免費(fèi)的天氣數(shù)據(jù)API，非常易于配置和使用，調(diào)用時(shí)只需傳入位置信息（城市名或者是地理坐標(biāo)）作為參數(shù)即可
通常要實(shí)現(xiàn)調(diào)用一個(gè)API需要如下這幾個(gè)步驟（每個(gè)步驟都有一堆公式化代碼）：

創(chuàng)建所需的模型類（必要時(shí)，添加上注解）。
實(shí)現(xiàn)請求—回應(yīng)管理的網(wǎng)絡(luò)層代碼，并帶錯(cuò)誤處理。
用后臺(tái)線程實(shí)現(xiàn)請求調(diào)用（一般是用異步任務(wù)的形式實(shí)現(xiàn)），用一個(gè)回調(diào)函數(shù)（Callback Function）來實(shí)現(xiàn)在UI線程上呈現(xiàn)回應(yīng)信息。
創(chuàng)建模型類

第一步我們可以依靠一些類似jsonschema2pojo的JSON-POJO生成工具（半）自動(dòng)化完成。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實(shí)現(xiàn)網(wǎng)絡(luò)調(diào)用

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

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

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

我們來看一個(gè)具體的調(diào)用示例（實(shí)際代碼中應(yīng)該把這個(gè)調(diào)用放到一個(gè)非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");
//...

怎么樣，很簡單吧，你只需要很少的代碼就實(shí)現(xiàn)了整個(gè)調(diào)用過程，這就是Retrofit的威力

用RxJava實(shí)現(xiàn)響應(yīng)式編程

現(xiàn)在我們就進(jìn)入第三步了：RxJava部分！我們這里示例將用它來實(shí)現(xiàn)異步的請求調(diào)用。
首先我們需要將前面創(chuàng)建的接口類，換為這個(gè)類：

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()這個(gè)方法，它調(diào)用了Observable.create()方法并向方法傳入一個(gè)Observable.OnSubscribeFunc的實(shí)現(xiàn)，以此得到一個(gè)Observable對象并返回。并且一旦Observable對象被訂閱（subscribed）后就會(huì)開始工作。Observable每次處理的結(jié)果都會(huì)當(dāng)作參數(shù)傳給onNext()方法。因?yàn)槲覀冞@里只是想實(shí)現(xiàn)網(wǎng)絡(luò)請求的并發(fā)調(diào)用，所以只需要讓每個(gè)Observable對象中調(diào)用一次請求即可。代碼最后調(diào)用onComplete()方法。這里的subscribeOn()方法很重要，它決定了程序?qū)⑦x用哪種線程。這里調(diào)用的是Schedulers.threadPoolForIO()，此線程用于優(yōu)化IO和網(wǎng)絡(luò)性能相關(guān)的工作。

最后一步是要實(shí)現(xiàn)這個(gè)API調(diào)用。下面的代碼實(shí)現(xiàn)了并發(fā)網(wǎng)絡(luò)請求，每個(gè)請求都使用不同的調(diào)用參數(shù)異步調(diào)用同一個(gè)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()方法將城市名稱數(shù)組轉(zhuǎn)化為一個(gè)observable對象，將數(shù)組里的字符串提供給不同的線程。然后mapMany()方法將會(huì)把前者提供的每一個(gè)字符串都轉(zhuǎn)化為observable對象（譯注：新對象包含的是weatherData對象數(shù)據(jù)）。這里的轉(zhuǎn)化通過調(diào)用ApiManager.getWeatherData()完成。

這里還是注冊在I/O線程池上。在Android系統(tǒng)上，如果需要把結(jié)果展示在UI上，就必須把數(shù)據(jù)發(fā)布給UI線程處理。因?yàn)槲覀冎溃贏ndroid上只有最原始的那個(gè)創(chuàng)建界面的線程才可以操作界面。這里只需要用observeOn()方法調(diào)用AndroidSchedulers.mainThread()即可。subscribe()方法的調(diào)用將觸發(fā)observable對象，我們可以在這里處理observable對象發(fā)出的結(jié)果。

這個(gè)示例展示了RxJava強(qiáng)大的功能。如果沒有Rx，我們需要?jiǎng)?chuàng)建N個(gè)線程去調(diào)用請求，然后通過異步方式把處理結(jié)果交給UI線程。使用Rx只需編寫很少的代碼就完成工作，使用它強(qiáng)大的功能創(chuàng)建、合并、過濾和轉(zhuǎn)化observable對象。

RxJava可以在開發(fā)安卓App時(shí)，作為一個(gè)強(qiáng)大的處理并發(fā)的工具使用。雖然要熟悉它還是需要一些時(shí)間，但是磨刀不誤砍柴工，一旦掌握了它，將給你帶來很大幫助。響應(yīng)式擴(kuò)展庫是個(gè)很好的想法，我們把它用于安卓程序的開發(fā)，已經(jīng)用了好幾個(gè)禮拜了（在不久的將來，我們產(chǎn)品的異步任務(wù)處理將完全基于它完成）。越是了解它，你就越會(huì)愛上它。

最新評(píng)論