前言
在第一篇講解okGo架構添加依賴支援時，還記得我們額外添加的兩個依賴嗎，一個okRx和一個okServer，這兩個均是基于okGo架構的擴充項目，其中okRx可以使請求結合RxJava一起使用，而okServer則提供了強大的下載下傳上傳功能，如斷點支援，多任務管理等，本篇我們主要講解okRx的使用。
OkRx主要功能
1，可以很完美結合RxJava做網絡請求
2，在使用上比Retrofit更簡單友善，門檻更低，靈活性更高
3，網絡請求和RxJava調用可以做成一條鍊試，寫法優雅
4，使用Converter接口，支援任意類型的資料自動解析
5，OkRx是擴充的OkGo，是以OkGo包含的所有功能和寫法，OkRx全部支援
Rx請求的寫法
1，擷取Observable對象
我們還是像正常使用OkGo的方式一樣，傳入我們需要請求的url，和我們需要的參數，寫法如下：
這裡有兩個特殊的方法：
.converter()：該方法參數是網絡請求的轉換器類執行個體，功能是将網絡請求的結果轉換成我們需要的資料類型。
.adapt()：該方法是方法傳回值的擴充卡，功能是将網絡請求的Call<T>對象，适配成我們需要的Observable<T>對象。
2. 調用Rx轉換代碼
上面兩步放在一起寫，如下：
上面的showLoading和dismissLoading方法是我們在BaseActivity中定義的顯示進度和隐藏進度的方法。
private ProgressDialog dialog;
public void showLoading() {
if (dialog != null && dialog.isShowing()) return;
dialog = new ProgressDialog(this);
dialog.requestWindowFeature(Window.FEATURE_NO_TITLE);
dialog.setCanceledOnTouchOutside(false);
dialog.setProgressStyle(ProgressDialog.STYLE_SPINNER);
dialog.setMessage("請求網絡中...");
dialog.show();
}
public void dismissLoading() {
if (dialog != null && dialog.isShowing()) {
dialog.dismiss();
}
}
addDisposable方法是取消請求時使用，見下面取消請求部分講解。
以上大概就是使用okGo結合RxJava使用的完整案例，其中有兩個地方作如下主要說明:
Converter詳解
convert接口實作類的作用是用來進行資料轉換的，即将網絡請求擷取到的資料轉換成我們泛型指定的對象，作用和retrofit中的gson轉換一樣，基于此種寫法我們就必須額外定義一個繼承Convert接口重寫convertResponse方法編寫解析轉換邏輯的實作類，因為.converter()參數為該轉換器類的一個執行個體，就不能像之前隻使用okGo那樣可以将該部分邏輯直接寫進callback實作類中的convertResponse方法之中而不去額外定義轉換器類了。
我們知道okGo架構預設提供了三種實作convertResponse轉換邏輯的三種convert：
1，StringConverter：按文本解析，解析的編碼依據服務端響應頭中的Content-Type中的編碼格式，自動進行編碼轉換，確定不出現亂碼。
2，BitmapConverter：如果請求的資料是圖檔，則可以使用該轉換器，該方法對圖檔進行了壓縮處理，確定不發生OOM。
3，FileConverter：如果要做檔案下載下傳，可以使用該轉換器，内部封裝了關于檔案下載下傳和進度回調的方法。
CallAdapter詳解
我們寫Rx請求的時候，最後調用了一個adapt()方法，這裡面需要一個CallAdapter擴充卡接口，那麼這個擴充卡是什麼東西呢，主要功能就是将我們得到的Call對象，适配成我們需要的Observable對象。
參考okrx2源碼的目錄結構如下：
在adapter包下，命名規則分為這麼幾組：
開頭的意思：
Observable開頭，表示傳回的對象是Observable對象
其他的：
結尾的意思：
Response結尾，表示傳回對象的泛型是Response
Body結尾，表示傳回對象的泛型是T，沒有外層的Response包裝
Result結尾，表示傳回對象的泛型是Result
OkRx使用緩存
其實關于adapt()方法還有個重載方法：
adapt(AdapterParam param，callAdapter<T,E> adapter)
可以額外傳遞一個AdapterParam對象作為第一個參數，我們都知道，okhttp有兩種請求方式，一種是同步請求，一種是異步請求，那麼這個參數就是控制把目前的Call對象适配成Observable時，是使用同步還是異步的方法，如果不傳該參數預設就是使用異步請求，AdapterParam對象隻有一個參數，如下：
如果是使用同步的方法，對應的，我們就使用的是okrx2源碼中的CallExecuteObservable對象來适配，如果是異步，那麼就使用CallEnqueueObservable對象來适配。他們有什麼差別的，其實隻有一個最大的差別，就是緩存的使用。
預設情況下，我們使用的是異步請求來适配的，使用方式和okgo使用緩存一樣，指定cacheKey和cacheMode就行了，如下：
那麼對于okrx緩存回調成功和網絡請求回調成功都是回調onNext()方法，那麼怎麼區分呢，回調的Response對象中有個isFromCache方法，就是表示目前回調是來自緩存還是網絡。
注意事項：
1，不同的緩存模式會導緻onNext()方法具有不同的回調次數，可能一次或者兩次
2， 如果使用同步方法來适配的話，緩存模式CacheMode.FIRST_CACHE_THEN_REQUEST是不生效的，異步才生效，原因是同步請求沒法傳回兩次資料。
取消請求
我們可以在基類BaseActivity中建立這麼兩個方法，把每一個請求的Disposable對象都交給由統一的CompositeDisposable對象去管理。
在Observer對象的onSubscibe方法中，把Disposable添加到CompositeDisposable中：
然後我們在目前activity界面銷毀onDestory方法中或者需要取消的地方調用取消請求的方法：
rxJava請求之map操作符的使用舉例：
map()操作符用于轉換一組資料。至于怎麼轉換，由你自己指定。比如，簡單的對資料進行加減乘除處理。又比如，将某一個類轉成另一個類。又比如，将一組User類對象(userName是User類的一個字段)，轉成一組userName對象。
@OnClick(R.id.jsonArrayRequest)
public void jsonArrayRequest(View view) {
OkGo.<LzyResponse<List<ServerModel>>>get(Urls.URL_JSONARRAY)//
.headers("aaa", "111")//
.params("bbb", "222")//
.converter(new JsonConvert<LzyResponse<List<ServerModel>>>() {})//
.adapt(new ObservableBody<LzyResponse<List<ServerModel>>>())//
.doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(@NonNull Disposable disposable) throws Exception {
showLoading();
}
})//
.map(new Function<LzyResponse<List<ServerModel>>, List<ServerModel>>() {
@Override
public List<ServerModel> apply(@NonNull LzyResponse<List<ServerModel>> response) throws Exception {
return response.data;
}
})//
.observeOn(AndroidSchedulers.mainThread())//
.subscribe(new Observer<List<ServerModel>>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
addDisposable(d);
}
@Override
public void onNext(@NonNull List<ServerModel> response) {
handleResponse(response);
}
@Override
public void onError(@NonNull Throwable e) {
e.printStackTrace(); //請求失敗
showToast("請求失敗");
handleError(null);
}
@Override
public void onComplete() {
dismissLoading();
}
});
}
Observer與Consumer的差別：他們都是充當觀察者角色，後者是簡化了前者，減去了許多回調接口。
Consumer是簡易版的Observer，他有多重重載，可以自定義你需要處理的資訊，我這裡調用的是隻接受onNext消息的方法，他隻提供一個回調接口accept，由于沒有onError和onCompete，無法再 接受到onError或者onCompete之後，實作函數回調。無法回調，并不代表不接收，他還是會接收到onCompete和onError之後做出預設操作，也就是監聽者（Consumer）不在接收 Observable發送的消息，下方的代碼測試了該效果：
final Consumer<String> consumer = new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.d("MainActivity", Thread.currentThread().getName() + " String:" + s);
}
};
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
Log.d("MainActivity", Thread.currentThread().getName() + "emit Hello");
e.onNext("Hello");
Log.d("MainActivity", Thread.currentThread().getName() + "emit Complete");
e.onComplete();
Log.d("MainActivity", Thread.currentThread().getName() + "emit World");
e.onNext("World");
}
});
參考部落格：https://blog.csdn.net/l_o_s/article/details/79409031