正文
這可能是最好的RxJava 2.x入門教程系列專欄
文章鏈接：
這可能是最好的RxJava 2.x 入門教程（完結(jié)版）【強(qiáng)力推薦】
這可能是最好的RxJava 2.x 入門教程（一）
這可能是最好的RxJava 2.x 入門教程（二）
這可能是最好的RxJava 2.x 入門教程（三）
這可能是最好的RxJava 2.x 入門教程（四）
這可能是最好的RxJava 2.x 入門教程（五）
GitHub 代碼同步更新：https://github.com/nanchen2251/RxJava2Examples
為了滿足大家的饑渴難耐，GitHub將同步更新代碼，主要包含基本的代碼封裝，RxJava 2.x所有操作符應(yīng)用場景介紹和實際應(yīng)用場景，后期除了RxJava可能還會增添其他東西，總之，GitHub上的Demo專為大家傾心打造。傳送門：https://github.com/nanchen2251/RxJava2Examples
一、前言
終于如愿來到讓我小伙伴們亢奮的 RxJava 2 使用場景舉例了，前面幾章中我們講解完了 RxJava 1.x 到 RxJava 2.x 的異同以及 RxJava 2.x 的各種操作符使用，如有疑問，歡迎點擊上方的鏈接進(jìn)入你想要的環(huán)節(jié)。
二、正題
1、簡單的網(wǎng)絡(luò)請求
想必大家都知道，很多時候我們在使用 RxJava 的時候總是和 Retrofit 進(jìn)行結(jié)合使用，而為了方便演示，這里我們就暫且采用 OkHttp3 進(jìn)行演示，配合 map，doOnNext ，線程切換進(jìn)行簡單的網(wǎng)絡(luò)請求：
1）通過 Observable.create() 方法，調(diào)用 OkHttp 網(wǎng)絡(luò)請求；
2）通過 map 操作符集合 gson，將 Response 轉(zhuǎn)換為 bean 類；
3）通過 doOnNext() 方法，解析 bean 中的數(shù)據(jù)，并進(jìn)行數(shù)據(jù)庫存儲等操作；
4) 調(diào)度線程，在子線程中進(jìn)行耗時操作任務(wù)，在主線程中更新 UI ；
5) 通過 subscribe()，根據(jù)請求成功或者失敗來更新 UI 。
Observable.create(new ObservableOnSubscribe<Response>() {
@Override public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
Builder builder = new Builder()
.url("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, MobileAddress>() {
@Override public MobileAddress apply(@NonNull Response response) throws Exception {
Log.e(TAG, "map 線程:" + Thread.currentThread().getName() + "\n"); if (response.isSuccessful()) {
ResponseBody body = response.body(); if (body != null) {
Log.e(TAG, "map:轉(zhuǎn)換前:" + response.body()); return new Gson().fromJson(body.string(), MobileAddress.class);
}
} return null;
}
}).observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<MobileAddress>() {
@Override public void accept(@NonNull MobileAddress s) throws Exception {
Log.e(TAG, "doOnNext 線程:" + Thread.currentThread().getName() + "\n");
mRxOperatorsText.append("\ndoOnNext 線程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "doOnNext: 保存成功：" + s.toString() + "\n");
mRxOperatorsText.append("doOnNext: 保存成功：" + s.toString() + "\n");
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<MobileAddress>() {
@Override public void accept(@NonNull MobileAddress data) throws Exception {
Log.e(TAG, "subscribe 線程:" + Thread.currentThread().getName() + "\n");
mRxOperatorsText.append("\nsubscribe 線程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "成功:" + data.toString() + "\n");
mRxOperatorsText.append("成功:" + data.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 線程:" + Thread.currentThread().getName() + "\n");
mRxOperatorsText.append("\nsubscribe 線程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "失?。?quot; + throwable.getMessage() + "\n");
mRxOperatorsText.append("失?。?quot; + throwable.getMessage() + "\n");
}
});
為了方便，我們后面的講解大部分采用開源的 Rx2AndroidNetworking 來處理，數(shù)據(jù)來源于天狗網(wǎng)等多個公共API接口。
mRxOperatorsText.append("RxNetworkActivity\n");
Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.build()
.getObjectObservable(MobileAddress.class)
.observeOn(AndroidSchedulers.mainThread()) // 為doOnNext() 指定在主線程，否則報錯
.doOnNext(new Consumer<MobileAddress>() {
@Override public void accept(@NonNull MobileAddress data) throws Exception {
Log.e(TAG, "doOnNext:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("\ndoOnNext:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG,"doOnNext:"+data.toString()+"\n");
mRxOperatorsText.append("doOnNext:"+data.toString()+"\n");
}
})
.map(new Function<MobileAddress, ResultBean>() {
@Override public ResultBean apply(@NonNull MobileAddress mobileAddress) throws Exception {
Log.e(TAG, "\n" );
mRxOperatorsText.append("\n");
Log.e(TAG, "map:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("map:"+Thread.currentThread().getName()+"\n" ); return mobileAddress.getResult();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<ResultBean>() {
@Override public void accept(@NonNull ResultBean data) throws Exception {
Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("\nsubscribe 成功:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG, "成功:" + data.toString() + "\n");
mRxOperatorsText.append("成功:" + data.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 失敗:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("\nsubscribe 失敗:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG, "失?。?quot;+ throwable.getMessage()+"\n" );
mRxOperatorsText.append("失?。?quot;+ throwable.getMessage()+"\n");
}
});
2、先讀取緩存，如果緩存沒數(shù)據(jù)再通過網(wǎng)絡(luò)請求獲取數(shù)據(jù)后更新UI
想必在實際應(yīng)用中，很多時候（對數(shù)據(jù)操作不敏感時）都需要我們先讀取緩存的數(shù)據(jù)，如果緩存沒有數(shù)據(jù)，再通過網(wǎng)絡(luò)請求獲取，隨后在主線程更新我們的UI。
concat 操作符簡直就是為我們這種需求量身定做。
concat 可以做到不交錯的發(fā)射兩個甚至多個 Observable 的發(fā)射事件，并且只有前一個 Observable 終止(onComplete) 后才會定義下一個 Observable。
利用這個特性，我們就可以先讀取緩存數(shù)據(jù)，倘若獲取到的緩存數(shù)據(jù)不是我們想要的，再調(diào)用 onComplete() 以執(zhí)行獲取網(wǎng)絡(luò)數(shù)據(jù)的Observable，如果緩存數(shù)據(jù)能應(yīng)我們所需，則直接調(diào)用 onNext() ，防止過度的網(wǎng)絡(luò)請求，浪費用戶的流量。
Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() {
@Override public void subscribe(@NonNull ObservableEmitter<FoodList> e) throws Exception {
Log.e(TAG, "create當(dāng)前線程:"+Thread.currentThread().getName() );
FoodList data = CacheManager.getInstance().getFoodListData(); // 在操作符 concat 中，只有調(diào)用 onComplete 之后才會執(zhí)行下一個 Observable
if (data != null){ // 如果緩存數(shù)據(jù)不為空，則直接讀取緩存數(shù)據(jù)，而不讀取網(wǎng)絡(luò)數(shù)據(jù)
isFromNet = false;
Log.e(TAG, "\nsubscribe: 讀取緩存數(shù)據(jù):" );
runOnUiThread(new Runnable() {
@Override public void run() {
mRxOperatorsText.append("\nsubscribe: 讀取緩存數(shù)據(jù):\n");
}
});
e.onNext(data);
}else {
isFromNet = true;
runOnUiThread(new Runnable() {
@Override public void run() {
mRxOperatorsText.append("\nsubscribe: 讀取網(wǎng)絡(luò)數(shù)據(jù):\n");
}
});
Log.e(TAG, "\nsubscribe: 讀取網(wǎng)絡(luò)數(shù)據(jù):" );
e.onComplete();
}
}
});
Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows",10+"")
.build()
.getObjectObservable(FoodList.class); // 兩個 Observable 的泛型應(yīng)當(dāng)保持一致
Observable.concat(cache,network)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<FoodList>() {
@Override public void accept(@NonNull FoodList tngouBeen) throws Exception {
Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName() ); if (isFromNet){
mRxOperatorsText.append("accept : 網(wǎng)絡(luò)獲取數(shù)據(jù)設(shè)置緩存: \n");
Log.e(TAG, "accept : 網(wǎng)絡(luò)獲取數(shù)據(jù)設(shè)置緩存: \n"+tngouBeen.toString作 者： 南 塵 
出 處： http://www.cnblogs.com/liushilin/
關(guān)于作者：專注于移動前端的項目開發(fā)。如有問題或建議，請多多賜教！歡迎加入Android交流群：118116509
版權(quán)聲明：本文版權(quán)歸作者和博客園共有，歡迎轉(zhuǎn)載，但未經(jīng)作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文鏈接。
特此聲明：所有評論和私信都會在第一時間回復(fù)。也歡迎園子的大大們指正錯誤，共同進(jìn)步?；蛘?a style="margin: 0px; padding: 0px; color: rgb(0, 0, 0); text-decoration-line: none;">直接私信我
聲援博主：如果您覺得文章對您有幫助，可以點擊文章下部【推薦】或側(cè)邊【關(guān)注】。您的鼓勵是作者堅持原創(chuàng)和持續(xù)寫作的最大動力！
http://www.cnblogs.com/liushilin/p/7110489.html
延伸閱讀
- 【知識整理】這可能是最好的RxJava 2.x 教程（完結(jié)版） 2017-07-21
- 【知識整理】這可能是最好的RxJava 2.x 入門教程（三） 2017-07-20
- 【知識整理】這可能是最好的RxJava 2.x 入門教程（二） 2017-07-06
- 【知識整理】這可能是最好的RxJava 2.x 入門教程（一） 2017-06-21
【知識整理】這可能是最好的RxJava 2.x 入門教程（三）