前言

上一篇文章，我們已經(jīng)了解了RxJava的相關(guān)概念以及基礎(chǔ)使用方式。這篇文章將會結(jié)合RxJava知識點與開發(fā)中遇到的實際場景進行講解，讓大家能夠通過這些例子更好的理解與掌握RxJava的用法與思想

數(shù)據(jù)請求

這是一個完整的數(shù)據(jù)請求例子，從緩存、網(wǎng)絡以及數(shù)據(jù)解析等。本例要實現(xiàn)的是若數(shù)據(jù)庫有對應的數(shù)據(jù)則直接使用緩存的數(shù)據(jù)，否則再通過網(wǎng)絡請求獲取數(shù)據(jù)，并且完成解析與緩存等操作

1.創(chuàng)建網(wǎng)絡請求Observable從網(wǎng)絡獲取數(shù)據(jù)
（1）Observable.create()調(diào)起網(wǎng)絡請求獲取數(shù)據(jù)
（2）map操作符將網(wǎng)絡響應映射為數(shù)據(jù)實體
（3）doOnNext操作符將數(shù)據(jù)存儲到數(shù)據(jù)庫
2.創(chuàng)建數(shù)據(jù)庫緩存讀取Observable讀取緩存
3.Observable.concat合并上述兩個Observable

        Observable net = Observable.create(new ObservableOnSubscribe<Response>() {
            @Override
            public void subscribe(ObservableEmitter<Response> e) throws Exception {
                Request.Builder builder = new Request.Builder()
                        .url("http://gank.io/api/data/Android/10/1")
                        .get();
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                e.onNext(response);
            }
        }).map(new Function<Response, GankIOAndroidEntity>() {
            @Override
            public GankIOAndroidEntity apply(Response response) throws Exception {
                ResponseBody body = response.body();
                if(body!=null){
                    return JSON.parseObject(body.toString(),GankIOAndroidEntity.class);
                }
                return null;
            }
        }).doOnNext(new Consumer<GankIOAndroidEntity>() {
            @Override
            public void accept(GankIOAndroidEntity gankIOAndroidEntity) throws Exception {
                saveDB();
            }
        });

        Observable cache = Observable.create(new ObservableOnSubscribe<GankIOAndroidEntity>() {
            @Override
            public void subscribe(ObservableEmitter<GankIOAndroidEntity> e) throws Exception {
                GankIOAndroidEntity entity = getFromDB();
                if(entity!=null){
                    e.onNext(entity);
                }else{
                    e.onComplete();
                }
            }
        });

        Observable.concat(cache,net)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<GankIOAndroidEntity>() {
                    @Override
                    public void accept(GankIOAndroidEntity gankIOAndroidEntity) throws Exception {
                        displayUI();
                    }
                });

當然還有一種情況，就是進入頁面時若有緩存先展示緩存，但不管是否有緩存都請求網(wǎng)絡數(shù)據(jù)并更新緩存。咱們只需要修改一下cache對象即可，不管緩存查詢結(jié)果如何，都執(zhí)行net操作

        Observable cache = Observable.create(new ObservableOnSubscribe<GankIOAndroidEntity>() {
            @Override
            public void subscribe(ObservableEmitter<GankIOAndroidEntity> e) throws Exception {
                GankIOAndroidEntity entity = getFromDB();
                if(entity!=null){
                    e.onNext(entity);
                }
                e.onComplete(); // 不管緩存查詢結(jié)果如何，都執(zhí)行net操作
            }
        });

網(wǎng)絡請求嵌套

我想大家肯定都遇到過類似這樣的情況，在某個接口請求完成后再進行下一個請求的調(diào)用。比如：先注冊成功后再調(diào)用登陸接口、先上傳頭像再保存資料、先更改數(shù)據(jù)再查詢數(shù)據(jù)等等例子

我們可以通過flatMap操作符來方便的實現(xiàn)上述的請求嵌套，擺脫天花亂墜的回調(diào)嵌套。復習一下flatMap的作用：FlatMap將一個發(fā)射數(shù)據(jù)的Observable變換為多個Observables，然后將它們發(fā)射的數(shù)據(jù)合并后放進一個單獨的Observable

        Observable.create(new ObservableOnSubscribe<CallBackEntity>() {
            @Override
            public void subscribe(ObservableEmitter<CallBackEntity> e) throws Exception {
                RequestBody body = new FormBody.Builder()
                        .add("username","lee")
                        .add("password","lee")
                        .build();
                Request.Builder builder = new Request.Builder()
                        .url("http://xxx/register")
                        .post(body);
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                CallBackEntity entity =
                        JSON.parseObject(response.body().toString(),CallBackEntity.class);
                e.onNext(entity);
            }
        }).flatMap(new Function<CallBackEntity, ObservableSource<CallBackEntity>>() {
            @Override
            public ObservableSource<CallBackEntity> apply(CallBackEntity callBackEntity) throws Exception {
                // register successful
                if(callBackEntity.getCode() == 0){
                    return Observable.create(new ObservableOnSubscribe<CallBackEntity>() {
                        @Override
                        public void subscribe(ObservableEmitter<CallBackEntity> e) throws Exception {
                            RequestBody body = new FormBody.Builder()
                                    .add("username","lee")
                                    .add("password","lee")
                                    .build();
                            Request.Builder builder = new Request.Builder()
                                    .url("http://xxx/login")
                                    .post(body);
                            Request request = builder.build();
                            Call call = new OkHttpClient().newCall(request);
                            Response response = call.execute();
                            CallBackEntity entity =
                                    JSON.parseObject(response.body().toString(),CallBackEntity.class);
                        }
                    });
                }
                return null;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<CallBackEntity>() {
                    @Override
                    public void accept(CallBackEntity callBackEntity) throws Exception {
                        displayUI();
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        // error
                    }
                });

數(shù)據(jù)合并

某個頁面的展示可能需要同時調(diào)用好多個接口來得到完整的數(shù)據(jù)，可以用zip合并數(shù)據(jù)

        Observable data1 = Observable.create(subscibe1);
        Observable data2 = Observable.create(subscibe2);
        Observable.zip(data1, data2, new BiFunction() {
            @Override
            public Object apply(Data1 o, Data2 o2) throws Exception {
                Data data = new Data(o,o2);
                return data;
            }
        }).subscribe(new Consumer() {
            @Override
            public void accept(Object o) throws Exception {
                displayUI();
            }
        });

按鈕防點擊

登陸按鈕、發(fā)送驗證碼按鈕等各種與接口請求或者邏輯操作關(guān)聯(lián)按鈕，防止多次點擊造成的非預期效果

        RxView.clicks(btn)
                .throttleFirst(1, TimeUnit.SECONDS)
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {

                    }
                });

復雜的數(shù)據(jù)變換

        Observable.just(dbData)
                .map(new Function<List<String>, Object>() {
                    @Override
                    public Object apply(List<String> strings) throws Exception {
                        return null;
                    }
                })
                .filter(new Predicate<Object>() {
                    @Override
                    public boolean test(Object o) throws Exception {
                        return false;
                    }
                }).distinct(new Function<Object, Object>() {
            @Override
            public Object apply(Object o) throws Exception {
                return null;
            }
        }).take(pageSize)
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {

                    }
                });

按鈕的增量監(jiān)聽

addTextChangedListener與setOnClickListener，相信大家都不陌生吧，不知道大家知道它們倆對于事件的監(jiān)聽有什么區(qū)別嗎？不了解的同學我解釋下，addTextChangedListener內(nèi)部維護了一個集合存儲設置的所有監(jiān)聽對象，當事件產(chǎn)生時會向這所有監(jiān)聽發(fā)送回調(diào)。而setOnClickListener內(nèi)部存儲了OnClickListener對象，用于保存設置的監(jiān)聽對象，也就是說只有一個。結(jié)果很明顯，addTextChangedListener是增量監(jiān)聽，setOnClickListener是覆蓋

那么有沒有辦法實現(xiàn)OnClickListener的增量監(jiān)聽呢？有的~
當然，類似的還有OnTouchListener、focusChanges、drags等等

        Observable<Object> shareClick = RxView.clicks(textView).share();
        Observable<MotionEvent> shareTouch = RxView.touches(textView).share();

        shareClick.subscribe(new Consumer<Object>() {
            @Override
            public void accept(Object o) throws Exception {
                // listener1
            }
        });

        shareClick.subscribe(new Consumer<Object>() {
            @Override
            public void accept(Object o) throws Exception {
                // listener2
            }
        });

獲取驗證碼倒計時

        verifyCodeObservable = RxView.clicks(mBt)
                .throttleFirst(SECOND, TimeUnit.SECONDS)
                .subscribeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Action1<Void>() {
                    @Override
                    public void call(Void aVoid) {
                        RxView.enabled(mBt).call(false);
                    }
                });

        verifyCodeObservable.subscribe(new Action1<Void>() {
                    @Override
                    public void call(Void aVoid) {
                        Observable.interval(1, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
                                .take(SECOND)
                                .subscribe(new Observer<Long>() {
                                    @Override
                                    public void onCompleted() {
                                        RxTextView.text(mBt).call("獲取驗證碼");
                                        RxView.enabled(mBt).call(true);
                                    }

                                    @Override
                                    public void onError(Throwable e) {
                                        Log.e(TAG, e.toString());
                                    }

                                    @Override
                                    public void onNext(Long aLong) {
                                        RxTextView.text(mBt).call("剩余" + (SECOND - aLong) + "秒");
                                    }
                                });
                    }
                });

必填項/表單驗證

        Observable<CharSequence> ObservableName = RxTextView.textChanges(mEtPhone);
        Observable<CharSequence> ObservablePassword = RxTextView.textChanges(mEtPassword);

        Observable.combineLatest(ObservableName, ObservablePassword, new Func2<CharSequence, CharSequence, Boolean>() {
            @Override
            public Boolean call(CharSequence phone, CharSequence password) {
                return isPhoneValid(phone.toString()) && isPasswordValid(password.toString());
            }
        }).subscribe(new Action1<Boolean>() {
            @Override
            public void call(Boolean aBoolean) {
                RxView.enabled(mBtLogin).call(aBoolean);
            }
        });

        RxView.clicks(mBtLogin)
                .throttleFirst(1, TimeUnit.SECONDS)
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Void>() {
                    @Override
                    public void call(Void aVoid) {
                        Toast.makeText(LoginActivity.this, "登錄成功！" ,Toast.LENGTH_SHORT).show();
                    }
                });

響應式SP

        SharedPreferences preferences = PreferenceManager.getDefaultSharedPreferences(this);
        RxSharedPreference rxSharedPreference = RxSharedPreference.create(preferences);
        Preferences<Boolean> choose = rxSharedPreference.getBoolean("choose",false);
        RxCompoundButton.checkedChanges(checkBox)
                .subscribe(rxSharedPreference.asAction());

RxJava各種擴展庫

rx-preferences -使SharedPreferences支持RxJava

RxAndroid -RxJava的Android拓展

RxLifecycle -幫助使用了RxJava的安卓應用控制生命周期

RxBinding -安卓UI控件的RxJava綁定API

storio -支持RxJava的數(shù)據(jù)庫

retrofit -支持RxJava的網(wǎng)絡請求庫

sqlbrite -支持RxJava的sqlite數(shù)據(jù)庫

RxPermissions -RxJava實現(xiàn)的Android運行時權(quán)限控制

reark -RxJava architecture library for Android

frodo -Android Library for Logging RxJava Observables and Subscribers.

總結(jié)

沒有做不到，只有想不到。加油~