這可能是最好的RxJava2.x入門教程（五） · 我的Android Wiki

## 前言終于如愿來到讓我小伙伴們亢奮的 RxJava 2 使用場景舉例了，前面幾章中我們講解完了 RxJava 1.x 到 RxJava 2.x 的異同以及 RxJava 2.x 的各種操作符使用，如有疑問，歡迎點擊上方的鏈接進入你想要的環節。 ## 正題 #### 簡單的網絡請求想必大家都知道，很多時候我們在使用 RxJava 的時候總是和 Retrofit 進行結合使用，而為了方便演示，這里我們就暫且采用 OkHttp3 進行演示，配合 `map`，`doOnNext` ，線程切換進行簡單的網絡請求： * 1）通過 Observable.create() 方法，調用 OkHttp 網絡請求； * 2）通過 map 操作符集合 gson，將 Response 轉換為 bean 類； * 3）通過 doOnNext() 方法，解析 bean 中的數據，并進行數據庫存儲等操作； * 4）調度線程，在子線程中進行耗時操作任務，在主線程中更新 UI ； * 5）通過 subscribe()，根據請求成功或者失敗來更新 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:轉換前:" + 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, "失敗：" + throwable.getMessage() + "\n"); mRxOperatorsText.append("失敗：" + throwable.getMessage() + "\n"); } }); ~~~ 為了方便，我們后面的講解大部分采用開源的 [Rx2AndroidNetworking](https://github.com/amitshekhariitbhu/Fast-Android-Networking) 來處理，數據來源于[天狗網](http://www.tngou.net/)等多個公共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, "失敗："+ throwable.getMessage()+"\n" ); mRxOperatorsText.append("失敗："+ throwable.getMessage()+"\n"); } }); ~~~ #### 先讀取緩存，如果緩存沒數據再通過網絡請求獲取數據后更新UI 想必在實際應用中，很多時候（對數據操作不敏感時）都需要我們先讀取緩存的數據，如果緩存沒有數據，再通過網絡請求獲取，隨后在主線程更新我們的 UI。 `concat` 操作符簡直就是為我們這種需求量身定做。 `concat` 可以做到不交錯的發射兩個甚至多個 `Observable` 的發射事件，并且只有前一個 `Observable` 終止( `onComplete()` ) 后才會定義下一個 `Observable`。利用這個特性，我們就可以先讀取緩存數據，倘若獲取到的緩存數據不是我們想要的，再調用 `onComplete()` 以執行獲取網絡數據的 `Observable`，如果緩存數據能應我們所需，則直接調用 `onNext()` ，防止過度的網絡請求，浪費用戶的流量。 ~~~ Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() { @Override public void subscribe(@NonNull ObservableEmitter<FoodList> e) throws Exception { Log.e(TAG, "create當前線程:"+Thread.currentThread().getName() ); FoodList data = CacheManager.getInstance().getFoodListData(); // 在操作符 concat 中，只有調用 onComplete 之后才會執行下一個 Observable if (data != null){ // 如果緩存數據不為空，則直接讀取緩存數據，而不讀取網絡數據 isFromNet = false; Log.e(TAG, "\nsubscribe: 讀取緩存數據:" ); runOnUiThread(new Runnable() { @Override public void run() { mRxOperatorsText.append("\nsubscribe: 讀取緩存數據:\n"); } }); e.onNext(data); }else { isFromNet = true; runOnUiThread(new Runnable() { @Override public void run() { mRxOperatorsText.append("\nsubscribe: 讀取網絡數據:\n"); } }); Log.e(TAG, "\nsubscribe: 讀取網絡數據:" ); e.onComplete(); } } }); Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list") .addQueryParameter("rows",10+"") .build() .getObjectObservable(FoodList.class); // 兩個 Observable 的泛型應當保持一致 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 : 網絡獲取數據設置緩存: \n"); Log.e(TAG, "accept : 網絡獲取數據設置緩存: \n"+tngouBeen.toString() ); CacheManager.getInstance().setFoodListData(tngouBeen); } mRxOperatorsText.append("accept: 讀取數據成功:" + tngouBeen.toString()+"\n"); Log.e(TAG, "accept: 讀取數據成功:" + tngouBeen.toString()); } }, new Consumer<Throwable>() { @Override public void accept(@NonNull Throwable throwable) throws Exception { Log.e(TAG, "subscribe 失敗:"+Thread.currentThread().getName() ); Log.e(TAG, "accept: 讀取數據失敗："+throwable.getMessage() ); mRxOperatorsText.append("accept: 讀取數據失敗："+throwable.getMessage()+"\n"); } }); ~~~ 有時候我們的緩存可能還會分為 memory 和 disk ，實際上都差不多，無非是多寫點 `Observable` ，然后通過 `concat` 合并即可。 #### 多個網絡請求依次依賴想必這種情況也在實際情況中比比皆是，例如用戶注冊成功后需要自動登錄，我們只需要先通過注冊接口注冊用戶信息，注冊成功后馬上調用登錄接口進行自動登錄即可。我們的 `flatMap` 恰好解決了這種應用場景，`flatMap` 操作符可以將一個發射數據的 `Observable` 變換為多個 `Observables` ，然后將它們發射的數據合并后放到一個單獨的 `Observable`，利用這個特性，我們很輕松地達到了我們的需求。 ~~~ Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list") .addQueryParameter("rows", 1 + "") .build() .getObjectObservable(FoodList.class) // 發起獲取食品列表的請求，并解析到FootList .subscribeOn(Schedulers.io()) // 在io線程進行網絡請求 .observeOn(AndroidSchedulers.mainThread()) // 在主線程處理獲取食品列表的請求結果 .doOnNext(new Consumer<FoodList>() { @Override public void accept(@NonNull FoodList foodList) throws Exception { // 先根據獲取食品列表的響應結果做一些操作 Log.e(TAG, "accept: doOnNext :" + foodList.toString()); mRxOperatorsText.append("accept: doOnNext :" + foodList.toString()+"\n"); } }) .observeOn(Schedulers.io()) // 回到 io 線程去處理獲取食品詳情的請求 .flatMap(new Function<FoodList, ObservableSource<FoodDetail>>() { @Override public ObservableSource<FoodDetail> apply(@NonNull FoodList foodList) throws Exception { if (foodList != null && foodList.getTngou() != null && foodList.getTngou().size() > 0) { return Rx2AndroidNetworking.post("http://www.tngou.net/api/food/show") .addBodyParameter("id", foodList.getTngou().get(0).getId() + "") .build() .getObjectObservable(FoodDetail.class); } return null; } }) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<FoodDetail>() { @Override public void accept(@NonNull FoodDetail foodDetail) throws Exception { Log.e(TAG, "accept: success ：" + foodDetail.toString()); mRxOperatorsText.append("accept: success ：" + foodDetail.toString()+"\n"); } }, new Consumer<Throwable>() { @Override public void accept(@NonNull Throwable throwable) throws Exception { Log.e(TAG, "accept: error :" + throwable.getMessage()); mRxOperatorsText.append("accept: error :" + throwable.getMessage()+"\n"); } }); ~~~ #### 結合多個接口的數據更新 UI 在實際應用中，我們極有可能會在一個頁面顯示的數據來源于多個接口，這時候我們的 `zip` 操作符為我們排憂解難。 `zip` 操作符可以將多個 `Observable` 的數據結合為一個數據源再發射出去。 ~~~ Observable<MobileAddress> observable1 = Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512") .build() .getObjectObservable(MobileAddress.class); Observable<CategoryResult> observable2 = Network.getGankApi() .getCategoryData("Android",1,1); Observable.zip(observable1, observable2, new BiFunction<MobileAddress, CategoryResult, String>() { @Override public String apply(@NonNull MobileAddress mobileAddress, @NonNull CategoryResult categoryResult) throws Exception { return "合并后的數據為：手機歸屬地："+mobileAddress.getResult().getMobilearea()+"人名："+categoryResult.results.get(0).who; } }).subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { Log.e(TAG, "accept: 成功：" + s+"\n"); } }, new Consumer<Throwable>() { @Override public void accept(@NonNull Throwable throwable) throws Exception { Log.e(TAG, "accept: 失敗：" + throwable+"\n"); } }); ~~~ #### 間隔任務實現心跳想必即時通訊等需要輪訓的任務在如今的 APP 中已是很常見，而 RxJava 2.x 的 `interval` 操作符可謂完美地解決了我們的疑惑。這里就簡單的意思一下輪訓。 ~~~ private Disposable mDisposable; @Override protected void doSomething() { mDisposable = Flowable.interval(1, TimeUnit.SECONDS) .doOnNext(new Consumer<Long>() { @Override public void accept(@NonNull Long aLong) throws Exception { Log.e(TAG, "accept: doOnNext : "+aLong ); } }) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<Long>() { @Override public void accept(@NonNull Long aLong) throws Exception { Log.e(TAG, "accept: 設置文本："+aLong ); mRxOperatorsText.append("accept: 設置文本："+aLong +"\n"); } }); } /** * 銷毀時停止心跳 */ @Override protected void onDestroy() { super.onDestroy(); if (mDisposable != null){ mDisposable.dispose(); } } ~~~ ## 后記姑且先講到這里吧，喜歡的小伙伴別忘了關注和點贊哦~ [https://github.com/nanchen2251/RxJava2Examples](https://github.com/nanchen2251/RxJava2Examples) 作者：南塵2251 鏈接：http://www.jianshu.com/p/81fac37430dd 來源：簡書著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。