前言
RxJava和Retrofit也火了一段時(shí)間了，不過(guò)最近一直在學(xué)習(xí)ReactNative和Node相關(guān)的姿勢(shì)，一直沒(méi)有時(shí)間研究這些新東西，最近有個(gè)項(xiàng)目準(zhǔn)備寫(xiě)，打算先用Android寫(xiě)一個(gè)Demo出來(lái)，卻發(fā)現(xiàn)Android的世界發(fā)生了天翻地覆的變化，EventBus和OKHttp啥的都不見(jiàn)了，RxJava和Retrofit是什么鬼？
好吧，到Github上耐著性子看過(guò)了RxJava和Retrofit的介紹和幾個(gè)Demo，原來(lái)Android的大神Jake Wharton為Retrofit這個(gè)項(xiàng)目貢獻(xiàn)了這么多的代碼，沒(méi)有道理不用了。
如果你對(duì)RxJava不熟悉請(qǐng)先看給 Android 開(kāi)發(fā)者的 RxJava 詳解這篇文章。
如果你對(duì)Retrofit不熟悉就先看Retrofit官網(wǎng)。
當(dāng)然也有很多RxJava與Retrofit的文章，但是我覺(jué)得很多大家都很糾結(jié)的功能都沒(méi)有被總結(jié)出來(lái)，所以才有了此篇文章。
歡迎大家拍磚。
接下來(lái)進(jìn)入正文，我是從下面幾個(gè)角度去思考RxJava與Retrofit結(jié)合的。
RxJava如何與Retrofit結(jié)合
相同格式的Http請(qǐng)求數(shù)據(jù)該如何封裝
相同格式的Http請(qǐng)求數(shù)據(jù)統(tǒng)一進(jìn)行預(yù)處理
如何取消一個(gè)Http請(qǐng)求 -- 觀察者之間的對(duì)決，Oberver VS Subscriber
一個(gè)需要ProgressDialog的Subscriber該有的樣子

1.RxJava如何與Retrofit結(jié)合
1.1 基本頁(yè)面
先扔出build.gradle
文件的內(nèi)容
dependencies { compile fileTree(dir: 'libs', include: ['*.jar']) testCompile 'junit:junit:4.12' compile 'com.android.support:appcompat-v7:23.2.0' compile 'io.reactivex:rxjava:1.1.0' compile 'io.reactivex:rxandroid:1.1.0' compile 'com.squareup.retrofit2:retrofit:2.0.0-beta4' compile 'com.squareup.retrofit2:converter-gson:2.0.0-beta4' compile 'com.squareup.retrofit2:adapter-rxjava:2.0.0-beta4' compile 'com.google.code.gson:gson:2.6.2' compile 'com.jakewharton:butterknife:7.0.1'}

也就是說(shuō)本文是基于RxJava1.1.0和Retrofit 2.0.0-beta4來(lái)進(jìn)行的。 添加rxandroid是因?yàn)閞xjava中的線程問(wèn)題。
下面先搭建一個(gè)基本的頁(yè)面，頁(yè)面很簡(jiǎn)單，先來(lái)看文件目錄結(jié)構(gòu)

目錄結(jié)構(gòu)

MainActivity.java的代碼如下：
package com.queen.rxjavaretrofitdemo.activity;import android.os.Bundle;import android.support.v7.app.AppCompatActivity;import android.widget.Button;import android.widget.TextView;import com.queen.rxjavaretrofitdemo.R;import butterknife.Bind;import butterknife.ButterKnife;import butterknife.OnClick;public class MainActivity extends AppCompatActivity { @Bind(R.id.click_me_BN) Button clickMeBN; @Bind(R.id.result_TV) TextView resultTV; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); ButterKnife.bind(this); } @OnClick(R.id.click_me_BN) public void onClick() { getMovie(); } //進(jìn)行網(wǎng)絡(luò)請(qǐng)求 private void getMovie(){ }}

注意不要忘記加網(wǎng)絡(luò)權(quán)限
<uses-permission android:name="android.permission.INTERNET"/>

1.2 只用Retrofit
我們準(zhǔn)備在getMovie方法中進(jìn)行網(wǎng)絡(luò)請(qǐng)求，我們先來(lái)看看只使用Retrofit是如何進(jìn)行的。
我們使用豆瓣電影的Top250做測(cè)試連接，目標(biāo)地址為
https://api.douban.com/v2/movie/top250?start=0&count=10

至于返回的數(shù)據(jù)格式，大家自己訪問(wèn)下鏈接就看到了，太長(zhǎng)就不放進(jìn)來(lái)了。
首先我們要根據(jù)返回的結(jié)果封裝一個(gè)Entity，暫命名為MovieEntity，代碼就不貼了。
接下來(lái)我們要?jiǎng)?chuàng)建一個(gè)接口取名為MovieService，代碼如下：
public interface MovieService { @GET("top250") Call<MovieEntity> getTopMovie(@Query("start") int start, @Query("count") int count);}

回到MainActivity之中，我們來(lái)寫(xiě)getMovie方法的代碼
//進(jìn)行網(wǎng)絡(luò)請(qǐng)求private void getMovie(){ String baseUrl = "https://api.douban.com/v2/movie/"; Retrofit retrofit = new Retrofit.Builder() .baseUrl(baseUrl) .addConverterFactory(GsonConverterFactory.create()) .build(); MovieService movieService = retrofit.create(MovieService.class); Call<MovieEntity> call = movieService.getTopMovie(0, 10); call.enqueue(new Callback<MovieEntity>() { @Override public void onResponse(Call<MovieEntity> call, Response<MovieEntity> response) { resultTV.setText(response.body().toString()); } @Override public void onFailure(Call<MovieEntity> call, Throwable t) { resultTV.setText(t.getMessage()); } });}

以上為沒(méi)有經(jīng)過(guò)封裝的、原生態(tài)的Retrofit寫(xiě)網(wǎng)絡(luò)請(qǐng)求的代碼。 我們可以封裝創(chuàng)建Retrofit和service部分的代碼，然后Activity用創(chuàng)建一個(gè)Callback作為參數(shù)給Call，這樣Activity中只關(guān)注請(qǐng)求的結(jié)果，而且Call有cancel方法可以取消一個(gè)請(qǐng)求，好像沒(méi)Rxjava什么事了，我覺(jué)得可以寫(xiě)到這就下班了~
接下來(lái)我們要面對(duì)的問(wèn)題是這樣的 如果我的Http返回?cái)?shù)據(jù)是一個(gè)統(tǒng)一的格式，例如
{ "resultCode": 0, "resultMessage": "成功", "data": {}}

我們?nèi)绾螌?duì)返回結(jié)果進(jìn)行一個(gè)統(tǒng)一的處理呢？
另外，我的ProgressDialog的show方法應(yīng)該在哪調(diào)用呢？看樣子只能在getMovie()這個(gè)方法里面調(diào)用了，換個(gè)地方發(fā)出請(qǐng)求就要在對(duì)應(yīng)的Listener里面寫(xiě)一遍show()的代碼，其實(shí)挺鬧心。
而且錯(cuò)誤請(qǐng)求我也想集中處理掉不要貼重復(fù)的代碼。
我們先來(lái)看結(jié)合了Rxjava之后，事情有沒(méi)有變化的可能。當(dāng)然即便是不用Rxjava，依舊能夠做很多的封裝，只是比較麻煩。
如需查看項(xiàng)目代碼 --> 代碼地址:
https://github.com/tough1985/RxjavaRetrofitDemo

選擇Tag -> step1
1.3 添加Rxjava
Retrofit本身對(duì)Rxjava提供了支持。
添加Retrofit對(duì)Rxjava的支持需要在Gradle文件中添加
compile 'com.squareup.retrofit2:adapter-rxjava:2.0.0-beta4'

當(dāng)然我們已經(jīng)添加過(guò)了。
然后在創(chuàng)建Retrofit的過(guò)程中添加如下代碼：
Retrofit retrofit = new Retrofit.Builder() .baseUrl(baseUrl) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .build();

這樣一來(lái)我們定義的service返回值就不在是一個(gè)Call了，而是一個(gè)Observable
重新定義MovieService
public interface MovieService { @GET("top250") Observable<MovieEntity> getTopMovie(@Query("start") int start, @Query("count") int count);}

getMovie方法改為：
//進(jìn)行網(wǎng)絡(luò)請(qǐng)求private void getMovie(){ String baseUrl = "https://api.douban.com/v2/movie/"; Retrofit retrofit = new Retrofit.Builder() .baseUrl(baseUrl) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .build(); MovieService movieService = retrofit.create(MovieService.class); movieService.getTopMovie(0, 10) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<MovieEntity>() { @Override public void onCompleted() { Toast.makeText(MainActivity.this, "Get Top Movie Completed", Toast.LENGTH_SHORT).show(); } @Override public void onError(Throwable e) { resultTV.setText(e.getMessage()); } @Override public void onNext(MovieEntity movieEntity) { resultTV.setText(movieEntity.toString()); } });}

這樣基本上就完成了Retrofit和Rxjava的結(jié)合，但是我知道你們當(dāng)然不會(huì)滿意的。
接下來(lái)我們把創(chuàng)建Retrofit的過(guò)程封裝一下，然后希望Activity創(chuàng)建Subscriber對(duì)象傳進(jìn)來(lái)。
如需查看項(xiàng)目代碼 --> 代碼地址:
https://github.com/tough1985/RxjavaRetrofitDemo

選擇Tag -> step2
1.4 將請(qǐng)求過(guò)程進(jìn)行封裝
創(chuàng)建一個(gè)對(duì)象HttpMethods
public class HttpMethods { public static final String BASE_URL = "https://api.douban.com/v2/movie/"; private static final int DEFAULT_TIMEOUT = 5; private Retrofit retrofit; private MovieService movieService; //構(gòu)造方法私有 private HttpMethods() { //手動(dòng)創(chuàng)建一個(gè)OkHttpClient并設(shè)置超時(shí)時(shí)間 OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder(); httpClientBuilder.connectTimeout(DEFAULT_TIMEOUT, TimeUnit.SECONDS); retrofit = new Retrofit.Builder() .client(httpClientBuilder.build()) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .baseUrl(BASE_URL) .build(); movieService = retrofit.create(MovieService.class); } //在訪問(wèn)HttpMethods時(shí)創(chuàng)建單例 private static class SingletonHolder{ private static final HttpMethods INSTANCE = new HttpMethods(); } //獲取單例 public static HttpMethods getInstance(){ return SingletonHolder.INSTANCE; } /** * 用于獲取豆瓣電影Top250的數(shù)據(jù) * @param subscriber 由調(diào)用者傳過(guò)來(lái)的觀察者對(duì)象 * @param start 起始位置 * @param count 獲取長(zhǎng)度 */ public void getTopMovie(Subscriber<MovieEntity> subscriber, int start, int count){ movieService.getTopMovie(start, count) .subscribeOn(Schedulers.io()) .unsubscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(subscriber); }}

用一個(gè)單例來(lái)封裝該對(duì)象，在構(gòu)造方法中創(chuàng)建Retrofit和對(duì)應(yīng)的Service。 如果需要訪問(wèn)不同的基地址，那么你可能需要?jiǎng)?chuàng)建多個(gè)Retrofit對(duì)象，或者干脆根據(jù)不同的基地址封裝不同的HttpMethod類(lèi)。
我們回頭再來(lái)看MainActivity中的getMovie方法：
private void getMovie(){ subscriber = new Subscriber<MovieEntity>() { @Override public void onCompleted() { Toast.makeText(MainActivity.this, "Get Top Movie Completed", Toast.LENGTH_SHORT).show(); } @Override public void onError(Throwable e) { resultTV.setText(e.getMessage()); } @Override public void onNext(MovieEntity movieEntity) { resultTV.setText(movieEntity.toString()); } }; HttpMethods.getInstance().getTopMovie(subscriber, 0, 10);}

其中subscriber是MainActivity的成員變量。
如需查看項(xiàng)目代碼 --> 代碼地址:
https://github.com/tough1985/RxjavaRetrofitDemo

選擇Tag -> step3
2.相同格式的Http請(qǐng)求數(shù)據(jù)該如何封裝
第二部分和第三部分我參考了知乎上的一個(gè)問(wèn)答： RxJava+Retrofit，在聯(lián)網(wǎng)返回后如何先進(jìn)行統(tǒng)一的判斷？ 不過(guò)沒(méi)有完整的示例，所以在這寫(xiě)一個(gè)完整的示例出來(lái)。
這個(gè)段落我們來(lái)聊一下有些Http服務(wù)返回一個(gè)固定格式的數(shù)據(jù)的問(wèn)題。 例如：
{ "resultCode": 0, "resultMessage": "成功", "data": {}}

大部分的Http服務(wù)可能都是這樣設(shè)置，resultCode和resultMessage的內(nèi)容相對(duì)比較穩(wěn)定，而data的內(nèi)容變化多端，72變都不一定夠變的，有可能是個(gè)User對(duì)象，也有可能是個(gè)訂單對(duì)象，還有可能是個(gè)訂單列表。 按照我們之前的用法，使用Gson轉(zhuǎn)型需要我們?cè)趧?chuàng)建subscriber對(duì)象是指定返回值類(lèi)型，如果我們對(duì)不同的返回值進(jìn)行封裝的話，那可能就要有上百個(gè)Entity了，看著明明是很清晰的結(jié)構(gòu)，卻因?yàn)閐ata的不確定性無(wú)奈了起來(lái)。
少年，不必?zé)?，?lái)來(lái)來(lái)~ 老衲賜你寶典葵花，老衲就是練了這個(gè)才出家。。。
我們可以創(chuàng)建一個(gè)HttpResult類(lèi)
public class HttpResult<T> { private int resultCode; private String resultMessage; private T data;}

如果data是一個(gè)User對(duì)象的話。那么在定義Service方法的返回值就可以寫(xiě)為
Observable<HttpResult<User>>

這樣一來(lái)HttpResult就相當(dāng)于一個(gè)包裝類(lèi)，將結(jié)果包裝了起來(lái)，但是在使用的時(shí)候要給出一個(gè)明確的類(lèi)型。
在上面的示例中，我也創(chuàng)建了一個(gè)HttpResult類(lèi)，用來(lái)模仿這個(gè)形式，將其中的Subject單獨(dú)封裝了起來(lái)。
public class HttpResult<T> { //用來(lái)模仿resultCode和resultMessage private int count; private int start; private int total; private String title; //用來(lái)模仿Data private T subjects;}

這樣泛型的時(shí)候就要寫(xiě)為：
Observable<HttpResult<List<Subject>>>

如需查看項(xiàng)目代碼 --> 代碼地址:
https://github.com/tough1985/RxjavaRetrofitDemo

選擇Tag -> step4
3.相同格式的Http請(qǐng)求數(shù)據(jù)統(tǒng)一進(jìn)行預(yù)處理
既然我們有了相同的返回格式，那么我們可能就需要在獲得數(shù)據(jù)之后進(jìn)行一個(gè)統(tǒng)一的預(yù)處理。
當(dāng)接收到了一個(gè)Http請(qǐng)求結(jié)果之后，由于返回的結(jié)構(gòu)統(tǒng)一為
{ "resultCode": 0, "resultMessage": "成功", "data": {}}

我們想要對(duì)resultCode和resultMessage先做一個(gè)判斷，因?yàn)槿绻鹯esultCode == 0
代表success，那么resultCode != 0
時(shí)data一般都是null。
Activity或Fragment對(duì)resultCode和resultMessage基本沒(méi)有興趣，他們只對(duì)請(qǐng)求狀態(tài)和data數(shù)據(jù)感興趣。
基于這種考慮，我們?cè)趓esultCode != 0
的時(shí)候，拋出個(gè)自定義的ApiException。這樣就會(huì)進(jìn)入到subscriber的onError中，我們可以在onError中處理錯(cuò)誤信息。
另外，請(qǐng)求成功時(shí)，需要將data數(shù)據(jù)轉(zhuǎn)換為目標(biāo)數(shù)據(jù)類(lèi)型傳遞給subscriber，因?yàn)?，Activity和Fragment只想拿到和他們真正相關(guān)的數(shù)據(jù)。
使用Observable的map方法可以完成這一功能。
在HttpMethods中創(chuàng)建一個(gè)內(nèi)部類(lèi)HttpResultFunc，代碼如下：
/** * 用來(lái)統(tǒng)一處理Http的resultCode,并將HttpResult的Data部分剝離出來(lái)返回給subscriber * * @param <T> Subscriber真正需要的數(shù)據(jù)類(lèi)型，也就是Data部分的數(shù)據(jù)類(lèi)型 */private class HttpResultFunc<T> implements Func1<HttpResult<T>, T>{ @Override public T call(HttpResult<T> httpResult) { if (httpResult.getResultCode() != 0) { throw new ApiException(httpResult.getResultCode()); } return httpResult.getData(); }}

然后我們的getTopMovie方法改為：
public void getTopMovie(Subscriber<List<Subject>> subscriber, int start, int count){ movieService.getTopMovie(start, count) .map(new HttpResultFunc<List<Subject>>()) .subscribeOn(Schedulers.io()) .unsubscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(subscriber);}

由于HttpResult中的泛型T就是我們希望傳遞給subscriber的數(shù)據(jù)類(lèi)型，而數(shù)據(jù)可以通過(guò)httpResult的getData方法獲得，這樣我們就處理了泛型問(wèn)題，錯(cuò)誤處理問(wèn)題，還有將請(qǐng)求數(shù)據(jù)部分剝離出來(lái)給subscriber
這樣我們只需要關(guān)注Data數(shù)據(jù)的類(lèi)型，而不必在關(guān)心整個(gè)過(guò)程了。
需要注意一點(diǎn)，就是在定義Service的時(shí)候，泛型是
HttpResult<User>//orHttpResult<List<Subject>>

而在定義Subscriber的時(shí)候泛型是 java User //or List<Subject>

不然你會(huì)得到一個(gè)轉(zhuǎn)型錯(cuò)誤。
如需查看項(xiàng)目代碼 --> 代碼地址:
https://github.com/tough1985/RxjavaRetrofitDemo

選擇Tag -> step5
代碼中我是用豆瓣數(shù)據(jù)模擬了HttpResult中的resultCode和resultMessage，與文檔中的代碼略有出入。
4.如何取消一個(gè)Http請(qǐng)求 -- 觀察者之間的對(duì)決，Observer VS Subscriber
4.1 取消一個(gè)Http請(qǐng)求
這一部分我們來(lái)聊一下關(guān)于取消Http請(qǐng)求的事情，已經(jīng)Oberver和Subscriber這兩個(gè)體位我們哪個(gè)更容易給我們G點(diǎn)。
如果沒(méi)有使用Rxjava，那么Service返回的是一個(gè)Call，而這個(gè)Call對(duì)象有一個(gè)cancel方法可以用來(lái)取消Http請(qǐng)求。那么用了Rxjava之后，如何來(lái)取消一個(gè)請(qǐng)求呢？因?yàn)榉祷刂凳且粋€(gè)Observable。我們能做的似乎只有解除對(duì)Observable對(duì)象的訂閱，其他的什么也做不了。
好在Retrofit已經(jīng)幫我們考慮到了這一點(diǎn)。 答案在RxJavaCallAdapterFactory這個(gè)類(lèi)的源碼中可以找到
static final class CallOnSubscribe<T> implements Observable.OnSubscribe<Response<T>> { private final Call<T> originalCall; CallOnSubscribe(Call<T> originalCall) { this.originalCall = originalCall; } @Override public void call(final Subscriber<? super Response<T>> subscriber) { // Since Call is a one-shot type, clone it for each new subscriber. final Call<T> call = originalCall.clone(); // Attempt to cancel the call if it is still in-flight on unsubscription. subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { call.cancel(); } })); try { Response<T> response = call.execute(); if (!subscriber.isUnsubscribed()) { subscriber.onNext(response); } } catch (Throwable t) { Exceptions.throwIfFatal(t); if (!subscriber.isUnsubscribed()) { subscriber.onError(t); } return; } if (!subscriber.isUnsubscribed()) { subscriber.onCompleted(); } }}

我們看到call方法中，給subscriber添加了一個(gè)Subscription對(duì)象，Subscription對(duì)象很簡(jiǎn)單，主要就是取消訂閱用的，如果你查看Subscriptions.create的源碼，發(fā)現(xiàn)是這樣的
public static Subscription create(final Action0 unsubscribe) { return BooleanSubscription.create(unsubscribe);}

利用了一個(gè)BooleanSubscription類(lèi)來(lái)創(chuàng)建一個(gè)Subscription，如果你點(diǎn)進(jìn)去看BooleanSubscription.create方法一切就清晰了，當(dāng)接觸綁定的時(shí)候，subscriber會(huì)調(diào)用Subscription的unsubscribe方法，然后觸發(fā)創(chuàng)建Subscription時(shí)候的傳遞進(jìn)來(lái)的Action0的call方法。RxJavaCallAdapterFactory幫我們給subscriber添加的是call.cancel()，
總結(jié)起來(lái)就是說(shuō)，我們?cè)贏ctivity或者Fragment中創(chuàng)建subscriber對(duì)象，想要取消請(qǐng)求的時(shí)候調(diào)用subscriber的unsubscribe方法就可以了。
對(duì)不起這一節(jié)有太多的Subscriber和Subscription以及Observer和Observable，老衲當(dāng)時(shí)看的時(shí)候也是不知道吐了多少次了，習(xí)慣了就好了。
4.2 為什么會(huì)提到Oberver
提到Observer的過(guò)程是這樣的。由于Subscriber一旦調(diào)用了unsubscribe方法之后，就沒(méi)有用了。且當(dāng)事件傳遞到onError或者onCompleted之后，也會(huì)自動(dòng)的解綁。這樣出現(xiàn)的一個(gè)問(wèn)題就是每次發(fā)送請(qǐng)求都要?jiǎng)?chuàng)建新的Subscriber對(duì)象。
這樣我們就把注意力放到了Observer，Observer本身是一個(gè)接口，他的特性是不管你怎么用，都不會(huì)解綁，為什么呢？因?yàn)樗麤](méi)有解綁的方法。所以就達(dá)到了復(fù)用的效果，一開(kāi)始我一直美滋滋的用Observer。事實(shí)上，如果你用的是Observer，在調(diào)用Observable對(duì)象的subscribe方法的時(shí)候，會(huì)自動(dòng)的將Observer對(duì)象轉(zhuǎn)換成Subscriber對(duì)象。
下面是源碼：
public final Subscription subscribe(final Observer<? super T> observer) { if (observer instanceof Subscriber) { return subscribe((Subscriber<? super T>)observer); } return subscribe(new Subscriber<T>() { @Override public void onCompleted() { observer.onCompleted(); } @Override public void onError(Throwable e) { observer.onError(e); } @Override public void onNext(T t) { observer.onNext(t); } });}

后來(lái)發(fā)現(xiàn)了問(wèn)題，
問(wèn)題1 無(wú)法取消，因?yàn)镺bserver沒(méi)有unsubscribe方法 問(wèn)題2 沒(méi)有onStart方法 這個(gè)一會(huì)聊

這兩個(gè)問(wèn)題是很痛苦的。所以，為了后面更好的高潮，我們還是選擇用Subscriber。
5.一個(gè)需要ProgressDialog的Subscriber該有的樣子
我們希望有一個(gè)Subscriber在我們每次發(fā)送請(qǐng)求的時(shí)候能夠彈出一個(gè)ProgressDialog，然后在請(qǐng)求接受的時(shí)候讓這個(gè)ProgressDialog消失，同時(shí)在我們?nèi)∠@個(gè)ProgressDialog的同時(shí)能夠取消當(dāng)前的請(qǐng)求，而我們只需要處理里面的數(shù)據(jù)就可以了。
我們先來(lái)創(chuàng)建一個(gè)類(lèi)，就叫ProgressSubscriber，讓他繼承Subscriber。
Subscriber給我們提供了onStart、onNext、onError、onCompleted四個(gè)方法。
其中只有onNext方法返回了數(shù)據(jù)，那我們自然希望能夠在onNext里面處理數(shù)據(jù)相關(guān)的邏輯。
onStart方法我們用來(lái)啟動(dòng)一個(gè)ProgressDialog。 onError方法我們集中處理錯(cuò)誤，同時(shí)也停止ProgressDialog onComplated方法里面停止ProgressDialog
其中我們需要解決兩個(gè)問(wèn)題
問(wèn)題1 onNext的處理 問(wèn)題2 cancel掉一個(gè)ProgressDialog的時(shí)候取消請(qǐng)求

我們先來(lái)解決問(wèn)題1
5.1處理onNext
我們希望這里能夠讓Activity或者Fragment自己處理onNext之后的邏輯，很自然的我們想到了用接口。問(wèn)題還是泛型的問(wèn)題，這里面我們必須指定明確的類(lèi)型。所以接口還是需要泛型。
我們先來(lái)定義一個(gè)接口，命名SubscriberOnNextListener
public interface SubscriberOnNextListener<T> { void onNext(T t);}

代碼很簡(jiǎn)單。再來(lái)看一下ProgressSubscriber現(xiàn)在的代碼
public class ProgressSubscriber<T> extends Subscriber<T> { private SubscriberOnNextListener mSubscriberOnNextListener; private Context context; public ProgressSubscriber(SubscriberOnNextListener mSubscriberOnNextListener, Context context) { this.mSubscriberOnNextListener = mSubscriberOnNextListener; this.context = context; } @Override public void onStart() { } @Override public void onCompleted() { Toast.makeText(context, "Get Top Movie Completed", Toast.LENGTH_SHORT).show(); } @Override public void onError(Throwable e) { Toast.makeText(context, "error:" + e.getMessage(), Toast.LENGTH_SHORT).show(); } @Override public void onNext(T t) { mSubscriberOnNextListener.onNext(t); }}

我知道傳Context不好，不過(guò)為了演示而已，大家可以自己封裝一下Toast。
MainActivity使用是這樣的：
先來(lái)定義一個(gè)SubscriberOnNextListener對(duì)象，可以在onCreate里面創(chuàng)建這個(gè)對(duì)象
private SubscriberOnNextListener getTopMovieOnNext;@Overrideprotected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); ButterKnife.bind(this); getTopMovieOnNext = new SubscriberOnNextListener<List<Subject>>() { @Override public void onNext(List<Subject> subjects) { resultTV.setText(subjects.toString()); } };}

getMovie方法這么寫(xiě)：
private void getMovie(){ HttpMethods.getInstance().getTopMovie( new ProgressSubscriber(getTopMovieOnNext, MainActivity.this), 0, 10);}

這樣Activity或Fragment就只需要關(guān)注拿到結(jié)果之后的邏輯了，其他的完全不用操心。
如需查看項(xiàng)目代碼 --> 代碼地址:
https://github.com/tough1985/RxjavaRetrofitDemo

選擇Tag -> step6
5.2處理ProgressDialog
我們希望當(dāng)cancel掉ProgressDialog的時(shí)候，能夠取消訂閱，也就取消了當(dāng)前的Http請(qǐng)求。 所以我們先來(lái)創(chuàng)建個(gè)接口來(lái)處理這件事情。
public interface ProgressCancelListener { void onCancelProgress();}

然后我們用ProgressSubscriber來(lái)實(shí)現(xiàn)這個(gè)接口，這樣ProgressSubscriber就有了一個(gè)onCancelProgress方法，在這里面取消訂閱。
@Overridepublic void onCancelProgress() { if (!this.isUnsubscribed()) { this.unsubscribe(); }}

然后我用了一個(gè)Handler來(lái)封裝了ProgressDialog。
public class ProgressDialogHandler extends Handler { public static final int SHOW_PROGRESS_DIALOG = 1; public static final int DISMISS_PROGRESS_DIALOG = 2; private ProgressDialog pd; private Context context; private boolean cancelable; private ProgressCancelListener mProgressCancelListener; public ProgressDialogHandler(Context context, ProgressCancelListener mProgressCancelListener, boolean cancelable) { super(); this.context = context; this.mProgressCancelListener = mProgressCancelListener; this.cancelable = cancelable; } private void initProgressDialog(){ if (pd == null) { pd = new ProgressDialog(context); pd.setCancelable(cancelable); if (cancelable) { pd.setOnCancelListener(new DialogInterface.OnCancelListener() { @Override public void onCancel(DialogInterface dialogInterface) { mProgressCancelListener.onCancelProgress(); } }); } if (!pd.isShowing()) { pd.show(); } } } private void dismissProgressDialog(){ if (pd != null) { pd.dismiss(); pd = null; } } @Override public void handleMessage(Message msg) { switch (msg.what) { case SHOW_PROGRESS_DIALOG: initProgressDialog(); break; case DISMISS_PROGRESS_DIALOG: dismissProgressDialog(); break; } }}

Handler接收兩個(gè)消息來(lái)控制顯示Dialog還是關(guān)閉Dialog。 創(chuàng)建Handler的時(shí)候我們需要傳入ProgressCancelListener的對(duì)象實(shí)例。
最后貼出ProgressSubscriber的完整代碼：
public class ProgressSubscriber<T> extends Subscriber<T> implements ProgressCancelListener{ private SubscriberOnNextListener mSubscriberOnNextListener; private ProgressDialogHandler mProgressDialogHandler; private Context context; public ProgressSubscriber(SubscriberOnNextListener mSubscriberOnNextListener, Context context) { this.mSubscriberOnNextListener = mSubscriberOnNextListener; this.context = context; mProgressDialogHandler = new ProgressDialogHandler(context, this, true); } private void showProgressDialog(){ if (mProgressDialogHandler != null) { mProgressDialogHandler.obtainMessage(ProgressDialogHandler.SHOW_PROGRESS_DIALOG).sendToTarget(); } } private void dismissProgressDialog(){ if (mProgressDialogHandler != null) { mProgressDialogHandler.obtainMessage(ProgressDialogHandler.DISMISS_PROGRESS_DIALOG).sendToTarget(); mProgressDialogHandler = null; } } @Override public void onStart() { showProgressDialog(); } @Override public void onCompleted() { dismissProgressDialog(); Toast.makeText(context, "Get Top Movie Completed", Toast.LENGTH_SHORT).show(); } @Override public void onError(Throwable e) { dismissProgressDialog(); Toast.makeText(context, "error:" + e.getMessage(), Toast.LENGTH_SHORT).show(); } @Override public void onNext(T t) { mSubscriberOnNextListener.onNext(t); } @Override public void onCancelProgress() { if (!this.isUnsubscribed()) { this.unsubscribe(); } }}

目前為止，就封裝完畢了。以上是我在用Rxjava和Retrofit過(guò)程中踩過(guò)的一些坑，最后整合出來(lái)的，由于沒(méi)有在實(shí)際的項(xiàng)目中跑過(guò)，有問(wèn)題的話希望能夠提出來(lái)大家討論一下，拍磚也歡迎。
現(xiàn)在我們?cè)賹?xiě)一個(gè)新的網(wǎng)絡(luò)請(qǐng)求，步驟是這樣的： 1. 在Service中定義一個(gè)新的方法。 2. 在HttpMethods封裝對(duì)應(yīng)的請(qǐng)求（代碼基本可以copy） 3. 創(chuàng)建一個(gè)SubscriberOnNextListener處理請(qǐng)求數(shù)據(jù)并刷新UI。
最后
如果你覺(jué)得寫(xiě)更改線程的代碼覺(jué)得也很煩的話，可以把訂閱這部分也封裝起來(lái)：
public void getTopMovie(Subscriber<List<Subject>> subscriber, int start, int count){ //原來(lái)的樣子// movieService.getTopMovie(start, count)// .map(new HttpResultFunc<List<Subject>>())// .subscribeOn(Schedulers.io())// .unsubscribeOn(Schedulers.io())// .observeOn(AndroidSchedulers.mainThread())// .subscribe(subscriber); //修改之后的樣子 Observable observable = movieService.getTopMovie(start, count) .map(new HttpResultFunc<List<Subject>>()); toSubscribe(observable, subscriber);}//添加線程管理并訂閱private void toSubscribe(Observable o, Subscriber s){ o.subscribeOn(Schedulers.io()) .unsubscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(s);}

讓你每次寫(xiě)一個(gè)請(qǐng)求的時(shí)候，寫(xiě)的代碼盡量少，更多的精力放在業(yè)務(wù)邏輯本身。
最后的最后
如果你的httpResult格式本身沒(méi)有問(wèn)題，但是data中的內(nèi)容是這樣的：
{ "resultCode": 0, "resultMessage": "成功", "data": {"user": {}, "orderArray": []}}

這樣的情況還能不能繼續(xù)使用這樣的框架呢？ 我的解決方法是封裝一個(gè)類(lèi)，把user和orderArray作為類(lèi)的屬性。 但是如果你的服務(wù)器一會(huì)data本身是一個(gè)完整的user數(shù)據(jù)，一會(huì)又是這樣： "data": {"user": {}, "orderArray": []}
那我覺(jué)得你有必要跟你的服務(wù)端好好聊聊了，請(qǐng)他吃頓飯和頓酒，大不了獻(xiàn)出菊花就是了。
但是如果服務(wù)已經(jīng)上線了！?。?br> 對(duì)不起，騷年......
老衲會(huì)在你墳前念300遍Thinking in java替你超度的~
希望你用Retrofit和Rxjava的新體位能夠享受到新的高潮。