RxJava操作符（五） ----過濾操作符

阿新 • • 發佈：2019-01-17

RxJava各型別操作符詳解如下：

過濾操作符：用於將Observable傳送的資料進行過濾和選擇。讓Observable返回我們所需要的資料。過濾操作符有buffer()，filter()，skip()，take()，skipLast()，takeLast()，throttleFirst()，distainctUntilChange()。

1、filter() 操作符

  /**
     * ========================filter() 操作符 ======================================
     * 
     * 對被觀察者傳送的事件做過濾操作。只有符合篩選條件的事件才會被觀察者所接收。
     * 
     * return true : 繼續傳送
     * 
     * return false ： 不傳送
     */ 

    public static void filter() {

        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        for (int i = 0; i < 3 
; i++) {
                            emitter.onNext(i);
                        }
                    }
                })
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        //return true : 繼續傳送 

                        //return false ： 不傳送
                        return integer != 2; // 本例子過濾了 等於2的情況
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG + "filter", String.valueOf(integer));
                    }
                });

    }

輸出如下：

2、distinct() 操作符

/**
     * ========================distinct 操作符 ======================================
     * 
     * 簡單地說就是去重。發射的資料包含重複的，會將重複的篩選掉。也就是，它只允許還沒有被髮射過的資料通過，被觀察者接收。發射完資料會自動呼叫onComplete()方法
     * 
     * y以下程式碼：觀察者只會接收到 ： 1,2,3,5 四個數值
     */
    public static void distinct() {
        Observable
                .just(1, 2, 3, 2, 3, 5)
                .distinct()
                .subscribe(
                        new Observer<Integer>() {
                            @Override
                            public void onSubscribe(Disposable d) {

                            }

                            @Override
                            public void onNext(Integer integer) {
                                Log.d(TAG + "distinct", String.valueOf(integer));
                            }

                            @Override
                            public void onError(Throwable e) {

                            }

                            @Override
                            public void onComplete() {
                                Log.d(TAG + "distinct", "onComplete");
                            }
                        });

    }

輸出如下：

3、distinctUntilChanged() 操作符

 /**
     * ========================distinctUntilChanged 操作符 ======================================
     * 
     * 過濾掉連續重複的事件
     */
    public static void distinctUntilChanged() {
        Observable.just(1, 2, 3, 1, 2, 3, 3, 4, 4)
                .distinctUntilChanged()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, String.valueOf(integer));
                    }
                });
    }

輸出如下：

4、skip()，skipLast()操作符


    /**
     * ========================skip，skipLast======================================
     *
     *  skip 操作符是把Observable發射的資料過濾點掉前n項。而take操作符只能接收前n項。另外還有skipLast和takeLast則是從後往前進行過濾.接收完會呼叫onComplete()方法
     *
     */

    /**
     * 以下程式碼輸出： 3,4,5
     */
    public static void skip() {
        Observable
                .just(1, 2, 3, 4, 5, 6, 7)
                .skip(2)  //過濾前2項
                .skipLast(3) //過濾後3項
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG + "skip", "根據順序過濾" + String.valueOf(integer));
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG + "skip", "onComplete");
                    }
                });
//-------------------------------------------------------------------------------------------------
        Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS)
                .skip(1, TimeUnit.SECONDS)   //過濾地1s傳送的資料
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.d(TAG + "skip", "根據事件過濾" + String.valueOf(aLong));
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });


    }

輸出如下：

5、 take() 操作符

/**
     * ====================== take 操作符===========================
     * 只能接收兩個事件
     */
    public static void take() {
        Observable
                .just(1, 2, 3, 4, 5, 6, 7)
                .take(2)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG + "skip", String.valueOf(integer));
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG + "skip", "onComplete");
                    }
                });
    }

輸出如下：

6、takeLast() 操作符

/**
     * =======================takeLast() 操作符 ======================
     * 
     * 只能接收被觀察者傳送的最後幾個事件
     */

    public static void takeLast() {
        Observable
                .just(1, 2, 3, 4, 5, 6, 7)
                .takeLast(2)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG + "skip", String.valueOf(integer));
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG + "skip", "onComplete");
                    }
                });
    }

輸出如下：

7、elementAt() 操作符


/**
     * ========================elementAt() 操作符 ======================================
     * 
     * 只發射第n項資料.
     *  一個引數和兩個引數時：
     *  elementAt(第n項)
     * elementAt(第n項，第N項不存在時預設值)
     * 
     *  n為負數時,報IndexOUtOfBoundExection。為正數但超過發射資料長度不會報異常會使用預設值代替
     */

    public static void elementAt() {

        Observable.range(1, 5)
                .elementAt(6, 10)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG + "elementAt", String.valueOf(integer));
                    }
                });
    }

輸出如下：

8、elementAtOrError()操作符

/**
     * ==============elementAtOrError()===================================================
     * 
     * 在elementAtError()的基礎上，當出現越界情況(當獲取位置的索引>事件序列的長度)，即丟擲異常
     */
    public static void elementAtOrError() {
        Observable.range(1, 5)
                .elementAtOrError(6)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG + "elementAtOrErr", String.valueOf(integer));
                    }
                });
    }

9、ignoreElements() 操作符

 /**
     * ========================ignoreElements() 操作符 ======================================
     * 
     * 不管發射的資料.只希望在它完成時和遇到錯誤時收到通知
     */
    public static void ignoreElements() {

        Observable
                .range(0, 10)
                .ignoreElements()
                .subscribe(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.d(TAG + "ignoreEles", "完成了");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG + "ignoreEles", "出錯了");
                    }
                });
    }

輸出如下：

10、ofType() 操作符

/**
     * ========================ofType() 操作符 ======================================
     * 
     * 通過資料的型別過濾資料，只發送指定型別資料。
     * 
     * 以下程式碼觀察者只接收到： 1,2
     */
    public static void ofType() {

        Observable.just("哈哈", 1, "嘻嘻", 2, 3.5)
                .ofType(Integer.class)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG + "ofType", String.valueOf(integer));
                    }
                });
    }

輸出如下：

11、throttleFirst()/throttleLast() 操作符

/**
     * ========================throttleFirst()/throttleLast() 操作符 ======================================
     * 
     * throttleFirst() 在某段時間內，只發送該段事件第一次事件
     * 
     * throttleLast()  在某段時間內，只發送該段事件最後一次事件
     */

    public static void throttleFirst() {
        Observable.interval(300, TimeUnit.MILLISECONDS) //每個0.3秒傳送一個事件
                .throttleFirst(1, TimeUnit.SECONDS) //只接收每秒內傳送的第一個資料
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d(TAG + "throttleFirst", String.valueOf(aLong));
                    }
                });
    }

輸出如下：

12、sample()

/**
     * ===================sample()=================================
     * 
     * 在某段時間內，只發送該段時間內最新(最後)1次事件
     * 
     * 與throttleLast類似
     */
    public static void sample() {
        Observable.interval(300, TimeUnit.MILLISECONDS) //每個0.3秒傳送一個事件
                .sample(1, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d(TAG + "sample", String.valueOf(aLong));
                    }
                });
    }

輸出如下：

13、firstElement()/lastElement()

/**
     * ========================firstElement()/lastElement()==========================
     * 
     * 選取第一個元素/最後一個元素
     */
    public static void firstElement() {
        Observable.just(1, 2, 3)
                .firstElement()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG + "firstElement", String.valueOf(integer));
                    }
                });
    }

輸出如下：

 public static void lastElement() {
        Observable
                .just(1, 2, 3)
                .lastElement()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG + "lastElement", String.valueOf(integer));
                    }
                });
    }

輸出如下：

RxJava各型別操作符詳解如下：

RxJava操作符（五） ----過濾操作符

RxJava各型別操作符詳解如下：過濾操作符：用於將Observable傳送的資料進行過濾和選擇。讓Observable返回我們所需要的資料。過濾操作符有buffer()，filter()，skip()，take()，skipLast(

初識RxJava（五）過濾類操作符

前言：前面已經記錄了四種操作符，分別是建立、延時、轉換、組合操作符，每天閒來無事看一篇。找個時間寫個demo 敲一下，然後就大體記住了，會用的第一步是記住，記住的第一步是認識。下面開始過濾操作符的筆記記錄；正文： 1、filter 操作符 1)、作用

初識RxJava（五）過濾類操作符

前言：前面已經記錄了四種操作符，分別是建立、延時、轉換、組合操作符，每天閒來無事看一篇。找個時間寫個demo 敲一下，然後就大體記住了，會用的第一步是記住，記住的第一步是認識。下面開始過濾操作符的筆記記錄；正文： 1、filter 操作符 1

RxJava（五） onErrorResumeNext操作符實現app與伺服器間token機制

RxJava系列文章目錄導讀：一、需求場景：在開發App的時候, 很多公司的提api介面, 請求的的時候都需要帶有token, 該token在使用者第一次啟動app或者登陸的時候去獲取. 以後的所有請求都需要帶該Token 如果token過期,

RxJava操作符（六）——功能操作符

一、功能操作符：輔助被觀察者（Observable）在傳送事件時實現一些功能性需求二、功能操作符按照使用功能，大致分類：訂閱：subscribe（）執行緒排程：subscribeOn（）、observeOn（）延遲：delay（）do操作：do（）錯誤處理：onErr

oracle學習筆記（五） SQL操作符

sel 算術 oracle學習 sca like 通配 union all 比較操作符 sele SQL操作符算術操作符：+加，-減，*乘，/除比較操作符： <,>,=,!=,<>,<=,>= 常用的判斷，<>和!=相同

Android RxJava操作符的學習---建立操作符--（無條件）網路請求輪詢

1. 需求場景 2. 功能說明本文將採用Get方法對金山詞霸API 按規定時間重複傳送網路請求，從而模擬輪詢需求實現採用 Gson 進行資料解析 3. 具體實現下面結合 Ret

RxJava（二） map操作符用法詳解

RxJava系列文章目錄導讀： 1 map操作符的作用 Returns an Observable that applies a specified function to each item emitted by the source Obse

RxJava2.0中flatMap操作符用法和原始碼分析（五）

flatMap基本使用 flatMap是變換操作符，使用一個指定的函式對原始Observable發射的每一項資料執行變換操作，這個函式返回一個本身也發射資料的Observable，然後flatMap合併這些Observable發射的資料，最後將合併後的結果當作

RxJava（一） create操作符的用法和原始碼分析

RxJava系列文章目錄導讀： 1 create操作符的基本使用顧名思義，Create操作符是用來建立一個Observable的。下面來看一個簡單的程式碼段： Observable.create(new Observable.OnSubscrib

RxJava（四） concatMap操作符用法詳解

RxJava系列文章目錄導讀： concatMap操作符的作用 concatMap操作符和flatMap操作符非常類似。如果對 flatMap操作符不是很瞭解可以點選連結去看看我的上一篇博文。下面是concatMap操作符的流程圖： conc

RxJava（三） flatMap操作符用法詳解

RxJava系列文章目錄導讀： flatMap操作符的作用官方文件解釋： Returns an Observable that emits items based on applying a function that you suppl

RxJava從入門到不離不棄（三）——轉換操作符

前面兩篇文章中我們介紹了RxJava的一些基本概念和RxJava最簡單的用法。從這一篇開始，我們開始聊聊RxJava中的操作符Operators。 RxJava中的操作符主要分成了三類：轉換類操作符(map flatMap concatMap flatM

weex 項目開發（五）自定義過濾函數和混合及自定義 Header 組件

定義 blog weex top ber slice ear notice earch 1.自定義過濾函數 src / filters / index.js /** * 自定義過濾函數 */ export function host (url) { if (

C#中如何利用操作符過載和轉換操作符（轉載）

操作符過載有的程式語言允許一個型別定義操作符應該如何操作型別的例項，比如string型別和int型別都過載了（==）和（+）等操作符，當編譯器發現兩個int型別的例項使用+操作符的時候，編譯器會生成把兩個整數加到一起的程式碼。當編譯器發現兩個string型別的例項使用+操作符的時候，編譯器會生成把兩個

安卓專案實戰之強大的網路請求框架okGo使用詳解（五）：擴充套件專案okRx，完美結合RxJava

前言在第一篇講解okGo框架新增依賴支援時，還記得我們額外新增的兩個依賴嗎，一個okRx和一個okServer，這兩個均是基於okGo框架的擴充套件專案，其中okRx可以使請求結合RxJava一起使用，而okServer則提供了強大的下載上傳功能，如斷點支援，多工管理等，本篇我們主要講

rxjs一句話描述一個操作符（1）

構造有一個函數式組播訂閱發布 ext push 訂閱模式之前一直在寫LINQ之類的東西，對於函數式的面向對象還是有一些功底，所以對於rxjs，感覺上不是很難，但是每次看完過幾天就忘，還是記一下筆記吧，好記性不如爛筆頭真不是蓋的。首先介紹幾個重要的概念。可以通

MongoDB 學習筆記（8）---$type 操作符

$type操作符是基於BSON型別來檢索集合中匹配的資料型別，並返回結果。 MongoDB 中可以使用的型別如下表所示：型別數字備註 Double 1 String 2 &n

MongoDB學習筆記（7）--- 條件操作符

描述條件操作符用於比較兩個表示式並從mongoDB集合中獲取資料。在本章節中，我們將討論如何在MongoDB中使用條件操作符。 MongoDB中條件操作符有： (>) 大於 - $gt (<) 小於 - $lt (>=) 大

.NET 官方驅動MongoDB.Driver的聚合管道Aggregate用法（二）之操作符的用法示例及細節

上一篇文章寫了如何去寫一個聚合管道的類以及例項化後的基本查詢實現。傳送門本文筆者想記錄在具體實現按日期統計資料進行Aggregate時，在編寫查詢條件時的一些容易忽略的小細節（筆者自己忽略的和遇到的問題）進行記錄，以備後用，錯誤之處，請留言指正。 1.時間問題

RxJava操作符（五） ----過濾操作符

RxJava各型別操作符詳解如下：

過濾操作符： 用於將Observable傳送的資料進行過濾和選擇。讓Observable返回我們所需要的資料。 過濾操作符有buffer()，filter()，skip()，take()，skipLast()，takeLast()，throttleFirst()，distainctUntilChange()。

RxJava各型別操作符詳解如下：

相關推薦

過濾操作符：用於將Observable傳送的資料進行過濾和選擇。讓Observable返回我們所需要的資料。過濾操作符有buffer()，filter()，skip()，take()，skipLast()，takeLast()，throttleFirst()，distainctUntilChange()。