【知識整理】這可能是最好的RxJava 2.x 入門教程（三）

 

 

這可能是最好的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所有操作符應用場景介紹和實際應用場景，后期除了RxJava可能還會增添其他東西，總之，GitHub上的Demo專為大家傾心打造。傳送門：https://github.com/nanchen2251/RxJava2Examples

 

一、前言

      年輕的老司機們，我這么勤的為大家分享，卻少有催更的，好吧。其實寫這個系列不是為了吸睛，那咱們繼續寫我們的 RxJava 2.x 的操作符。

二、正題

7、distinct

       這個操作符非常的簡單、通俗、易懂，就是簡單的去重嘛，我甚至都不想貼代碼，但人嘛，總得持之以恆。

Observable.just(1, 1, 1, 2, 2, 3, 4, 5)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("distinct : " + integer + "\n");
                        Log.e(TAG, "distinct : " + integer + "\n");
                    }
                });

輸出：

Log 日志顯而易見，我們在經過dinstinct() 后接收器接收到的事件只有1,2,3,4,5了。

8、Filter

      信我，Filter 你會很常用的，它的作用也很簡單，過濾器嘛。可以接受一個參數，讓其過濾掉不符合我們條件的值

Observable.just(1, 20, 65, -5, 7, 19)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(@NonNull Integer integer) throws Exception {
                        return integer >= 10;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                mRxOperatorsText.append("filter : " + integer + "\n");
                Log.e(TAG, "filter : " + integer + "\n");
            }
        });

輸出：

可以看到，我們過濾器舍去了小於10的值，所以最好的輸出只有20,65,19。

 

9、buffer

 buffer 操作符接受兩個參數，buffef(count,skip)，作用是將 Observable 中的數據按 skip (步長) 分成最大不超過 count 的 buffer ，然后生成一個 Observable 。也許你還不太理解，我們可以通過我們的示例圖和示例代碼來進一步深化它。

Observable.just(1, 2, 3, 4, 5)
                .buffer(3, 2)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(@NonNull List<Integer> integers) throws Exception {
                        mRxOperatorsText.append("buffer size : " + integers.size() + "\n");
                        Log.e(TAG, "buffer size : " + integers.size() + "\n");
                        mRxOperatorsText.append("buffer value : ");
                        Log.e(TAG, "buffer value : " );
                        for (Integer i : integers) {
                            mRxOperatorsText.append(i + "");
                            Log.e(TAG, i + "");
                        }
                        mRxOperatorsText.append("\n");
                        Log.e(TAG, "\n");
                    }
                });

輸出：

如圖，我們把1,2,3,4,5依次發射出來，經過buffer 操作符，其中參數 count 為3， skip 為2，而我們的輸出 依次是 123，345，5。顯而易見，我們 buffer 的第一個參數是count，代表最大取值，在事件足夠的時候，一般都是取count個值，然后每次跳過skip個事件。其實看 Log 日志，我相信大家都明白了。

 

10、timer

timer 很有意思，相當於一個定時任務。在1.x 中它還可以執行間隔邏輯，但在2.x中此功能被交給了 interval，下一個會介紹。但需要注意的是，timer 和 interval 均默認在新線程。

mRxOperatorsText.append("timer start : " + TimeUtil.getNowStrTime() + "\n");
        Log.e(TAG, "timer start : " + TimeUtil.getNowStrTime() + "\n");
        Observable.timer(2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()) // timer 默認在新線程，所以需要切換回主線程
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        mRxOperatorsText.append("timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                        Log.e(TAG, "timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                    }
                });

輸出：

顯而易見，當我們兩次點擊按鈕觸發這個事件的時候，接收被延遲了2秒。

 

11、interval

如同我們上面可說，interval 操作符用於間隔時間執行某個操作，其接受三個參數，分別是第一次發送延遲，間隔時間，時間單位。

 

mRxOperatorsText.append("interval start : " + TimeUtil.getNowStrTime() + "\n");
        Log.e(TAG, "interval start : " + TimeUtil.getNowStrTime() + "\n");
        Observable.interval(3,2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()) // 由於interval默認在新線程，所以我們應該切回主線程
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        mRxOperatorsText.append("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                        Log.e(TAG, "interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                    }
                });

輸出：

如同 Log 日志一樣，第一次延遲了3秒后接收到，后面每次間隔了2秒。

然而，心細的小伙伴可能會發現，由於我們這個是間隔執行，所以當我們的Activity都銷毀的時候，實際上這個操作還依然在進行，所以，我們得花點小心思讓我們在不需要它的時候干掉它。查看源碼發現，我們subscribe(Cousumer<? super T> onNext)返回的是Disposable，我們可以在這上面做文章。

@Override
    protected void doSomething() {
        mRxOperatorsText.append("interval start : " + TimeUtil.getNowStrTime() + "\n");
        Log.e(TAG, "interval start : " + TimeUtil.getNowStrTime() + "\n");
        mDisposable = Observable.interval(3, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()) // 由於interval默認在新線程，所以我們應該切回主線程
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        mRxOperatorsText.append("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                        Log.e(TAG, "interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                    }
                });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mDisposable != null && !mDisposable.isDisposed()) {
            mDisposable.dispose();
        }
    }

哈哈，再次驗證，解決了我們的疑惑。

 

12、doOnNext

其實覺得 doOnNext 應該不算一個操作符，但考慮到其常用性，我們還是咬咬牙將它放在了這里。它的作用是讓訂閱者在接收到數據之前干點有意思的事情。假如我們在獲取到數據之前想先保存一下它，無疑我們可以這樣實現。

Observable.just(1, 2, 3, 4)
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("doOnNext 保存 " + integer + "成功" + "\n");
                        Log.e(TAG, "doOnNext 保存 " + integer + "成功" + "\n");
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                mRxOperatorsText.append("doOnNext :" + integer + "\n");
                Log.e(TAG, "doOnNext :" + integer + "\n");
            }
        });

輸出：

 

13、skip

 skip 很有意思，其實作用就和字面意思一樣，接受一個 long 型參數 count ，代表跳過 count 個數目開始接收。

 Observable.just(1,2,3,4,5)
                .skip(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("skip : "+integer + "\n");
                        Log.e(TAG, "skip : "+integer + "\n");
                    }
                });

輸出：

 

 

14、take

take，接受一個 long 型參數 count ，代表至多接收 count 個數據。

 Flowable.fromArray(1,2,3,4,5)
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("take : "+integer + "\n");
                        Log.e(TAG, "accept: take : "+integer + "\n" );
                    }
                });

輸出：

 

15、just

 just，沒什么好說的，其實在前面各種例子都說明了，就是一個簡單的發射器依次調用onNext()方法。

Observable.just("1", "2", "3")
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        mRxOperatorsText.append("accept : onNext : " + s + "\n");
                        Log.e(TAG,"accept : onNext : " + s + "\n" );
                    }
                });

 輸出：

 

三、寫在最后

      好吧，本節先講到這里，下節我們還是繼續講簡單的操作符，雖然我們的教程比較枯燥，現在也不那么受人關注，但后面的系列我相信大家一定會非常喜歡的，我們下期再見！

     代碼全部同步到GitHub：https://github.com/nanchen2251/RxJava2Examples