RxJava学习指南(六)——过滤操作符

RxJava学习指南

版本：4
整理：陈小默

七、过滤操作符

7.1 Debounce

限流操作符的作用是过滤发射频率过快的数据，该操作符会记录每一个数据发射的时间，如果在安全时间限度内没有其他数据发射，则发射此数据，否则忽略此数据并监听最近发射的数据。

RxJava将这个操作符实现为throttleWithTimeout和debounce。

注意：onCompleted通知不会触发限流。

7.1.1 throttleWithTimeout

throtleWithTimeout/debounce的一个变体根据你指定的时间间隔进行限流，时间单位通过TimeUnit参数指定。

这种操作符默认在computation调度器上执行，但是你可以通过第三个参数指定。

    Observable.create<Int> {
        with(it) {
            try {
                onNext(0)
                Thread.sleep(50L)
                for (i in 0 until 10) onNext(i)
                Thread.sleep(60L)
                onNext(10)
            } catch (e: Exception) {
                onError(e)
            }
            onCompleted()
        }
    }.debounce(50L, TimeUnit.MILLISECONDS)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

Item: 0
Item: 9
Item: 10
Sequence complete

• Javadoc: throttleWithTimeout(long,TimeUnit) and debounce(long,TimeUnit)
• Javadoc: throttleWithTimeout(long,TimeUnit,Scheduler) and debounce(long,TimeUnit,Scheduler)

7.1.2 debounce

这里提供一个新的限流计时方式，既使用Observable充当计时器。我们可以在每个数据发射时创建一个Observable，如果在这个Observable完成之前，有新的数据发射出来，那么就直接丢弃当前的数据项，否则在Observable完成之时发射数据。

debounce的这个变体默认不在任何特定的调度器上执行。

    Observable.create<Int> {
        with(it) {
            try {
                onNext(0)
                Thread.sleep(50L)
                for (i in 0 until 10) onNext(i)
                Thread.sleep(60L)
                onNext(10)
            } catch (e: Exception) {
                onError(e)
            }
            onCompleted()
        }
    }.debounce { Observable.timer(50L, TimeUnit.MILLISECONDS) }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

7.2 Distinct

抑制（过滤掉）重复的数据项

Distinct的过滤规则是：只允许还没有发射过的数据项通过。

在某些实现中，有一些变体允许你调整判定两个数据不同(distinct)的标准。还有一些实现只比较一项数据和它的直接前驱，因此只会从序列中过滤掉连续重复的数据。

7.2.1 distinct()

RxJava将这个操作符实现为distinct函数。

示例代码

    Observable.just(1, 2, 3, 2, 1, 4, 5).distinct()
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

输出

Item: 1
Item: 2
Item: 3
Item: 4
Item: 5
Sequence complete

• Javadoc: distinct()

7.2.2 distinct(Func1)

这个操作符有一个变体接受一个函数。这个函数根据原始Observable发射的数据项产生一个Key，然后，比较这些Key。如果两个数据的key相同，则只保留最先到达的数据。

    Observable.just(1, 2, 3.0F, 4.0F, 5.0, 6.0, true, false)
            .distinct {
                when (it) {
                    is Int -> "I"
                    is Float -> "F"
                    is Double -> "D"
                    is Boolean -> "B"
                    else -> "A"
                }
            }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

Lambda：给接收到的数据创建一个key

• Javadoc: distinct(Func1)

7.2.3 distinctUntilChanged

RxJava还是实现了一个distinctUntilChanged操作符。它只判定一个数据和它的直接前驱是否是不同的。

    Observable.just(1, 2, 2, 1, 2, 2)
            .distinctUntilChanged()
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

7.3.4 distinctUntilChanged(Func1)

和distinct(Func1)一样，根据一个函数产生的Key判定两个相邻的数据项是不是不同的。

    Observable.just(1, 2.0F, 3.0F, 4, 5.0F, 6.0F)
            .distinctUntilChanged {
                when (it) {
                    is Int -> "I"
                    is Float -> "F"
                    else -> "A"
                }
            }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

Lambda：给接收到的数据创建一个key

Item: 1
Item: 2.0
Item: 4
Item: 5.0
Sequence complete

• Javadoc: distinctUntilChanged(Func1)

distinct和distinctUntilChanged默认不在任何特定的调度器上执行。

7.4 ElementAt

只发射第N项数据

ElementAt操作符获取原始Observable发射的数据序列指定索引位置的数据项，然后当做自己的唯一数据发射。

RxJava将这个操作符实现为elementAt，给它传递一个基于0的索引值，它会发射原始Observable数据序列对应索引位置的值，如果你传递给elementAt的值为5，那么它会发射第六项的数据。

如果你传递的是一个负数，或者原始Observable的数据项数小于index+1，将会抛出一个IndexOutOfBoundsException异常。

    Observable.just(0, 1, 2, 3, 4, 5)
            .elementAt(2)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc: elementAt(int)

7.4.1 elementAtOrDefault

RxJava还实现了elementAtOrDefault操作符。与elementAt的区别是，如果索引值大于数据项数，它会发射一个默认值（通过额外的参数指定），而不是抛出异常。但是如果你传递一个负数索引值，它仍然会抛出一个IndexOutOfBoundsException异常。

    Observable.just(0, 1, 2, 3)
            .elementAtOrDefault(6, -1)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc: elementAtOrDefault(int,T)

elementAt和elementAtOrDefault默认不在任何特定的调度器上执行。

7.5 Filter

过滤操作符，只发射满足条件的数据。

RxJava将这个操作符实现为filter函数。

示例代码

    Observable.just(0, 1, 2, 3, 4, 5, 6)
            .filter { 0 == it % 2 }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

Lambda：测试结果

Item: 0
Item: 2
Item: 4
Item: 6
Sequence complete

filter默认不在任何特定的调度器上执行。

• Javadoc: filter(Func1)

7.5.1 ofType

ofType是filter操作符的一个特殊形式。它过滤一个Observable只返回指定类型的数据。

    Observable.just(0, 1, "hello", 3, "world", 5, 6)
            .ofType(String::class.java)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

ofType默认不在任何特定的调度器上指定。

• Javadoc: ofType(Class)

7.6 First

只发射第一项（或者满足某个条件的第一项）数据

在某些实现中，First没有实现为一个返回Observable的过滤操作符，而是实现为一个在当时就发射原始Observable指定数据项的阻塞函数。在这些实现中，如果你想要的是一个过滤操作符，最好使用Take(1)或者ElementAt(0)。

在一些实现中还有一个Single操作符。它的行为与First类似，但为了确保只发射单个值，它会等待原始Observable终止（否则，不是发射那个值，而是以一个错误通知终止）。你可以使用它从原始Observable获取第一项数据，而且也确保只发射一项数据。

在RxJava中，这个操作符被实现为first，firstOrDefault和takeFirst。

可能容易混淆，BlockingObservable也有名叫first和firstOrDefault的操作符，它们会阻塞并返回值，不是立即返回一个Observable。

还有几个其它的操作符执行类似的功能。

7.5.1 过滤操作符

只发射第一个数据，使用没有参数的first操作符。

示例代码

    Observable.just(1, 2, 3)
            .first()
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc: first()

7.5.2 first(Func1)

发射第一个满足条件的数据。

    Observable.just(1, 2, 3)
            .first { 0 == it % 2 }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc: first(Func1)

7.5.3 firstOrDefault

firstOrDefault与first类似，如果没有满足条件的数据时就发射一个默认值。

    Observable.empty<Int>()
            .firstOrDefault(-1)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc: firstOrDefault(T)

7.5.4 firstOrDefault(Func1)

发射这个函数判定为true的第一项数据，如果没有满足条件的数据时就发射一个默认值。

    Observable.just(1, 3, 5, 7, 9)
            .firstOrDefault(-1, { 0 == it % 2 })
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc firstOrDefault(T, Func1)

7.5.5 takeFirst

takeFirst与first类似，除了这一点：如果原始Observable没有发射任何满足条件的数据，first会抛出一个NoSuchElementException，takeFist会返回一个空的Observable（不调用onNext()但是会调用onCompleted）。

    Observable.just(1, 3, 5, 7, 9)
            .takeFirst { 0 == it % 2 }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

Sequence complete

• Javadoc: takeFirst(Func1)

7.5.6 single

该操作符用来判断当前Observable是否只发射一个数据，如果是，则正常发射数据，否则抛出一个NoSuchElementException。

    Observable.from(arr)
            .single()
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

当val arr = arrayOf(1)时

Item: 1
Sequence complete

当val arr = emptyArray<Int>()时

Error: Sequence contains no elements

当val arr = arrayOf(1,2,3)时

Error: Sequence contains too many elements

• Javadoc: single()

7.5.7 single(Func1)

single的变体接受一个谓词函数，发射满足条件的单个值，如果不是正好只有一个数据项满足条件，会以错误通知终止。

    Observable.from(arr)
            .single { 0 == it % 2 }
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

val arr = arrayOf(1, 3, 5)时，打印

Error: Sequence contains no elements

val arr = arrayOf(1, 2, 3, 5)，打印

Item: 2
Sequence complete

val arr = arrayOf(1, 2, 3, 4, 5)，打印

Error: Sequence contains too many elements

• Javadoc: single(Func1)

7.5.8 singleOrDefault

和firstOrDefault类似，但是如果原始Observable发射超过一个的数据，会以错误通知终止。

    Observable.from(arr)
            .singleOrDefault(-1)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

val arr = emptyArray<Int>()，打印

Item: -1
Sequence complete

val arr = arrayOf(1, 2, 3, 4, 5)，打印

Error: Sequence contains too many elements

• Javadoc: singleOrDefault(T)

7.5.9 singleOrDefault(T,Func1)

和firstOrDefault(T, Func1)类似，如果没有数据满足条件，返回默认值；如果有多个数据满足条件，以错误通知终止。

    Observable.from(arr)
            .singleOrDefault(-1, { it % 2 == 0 })
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

• Javadoc: singleOrDefault(Func1,T)

first系列的这几个操作符默认不在任何特定的调度器上执行。

7.6 IgnoreElements

不发射任何数据，只发射Observable的终止通知(正常终止，或者错误终止)。

• Javadoc: ignoreElements()

ignoreElements默认不在任何特定的调度器上执行。

7.7 Last

只发射最后一项（或者满足某个条件的最后一项）数据

在某些实现中，Last没有实现为一个返回Observable的过滤操作符，而是实现为一个在当时就发射原始Observable指定数据项的阻塞函数。在这些实现中，如果你想要的是一个过滤操作符，最好使用TakeLast(1)。

在RxJava中的实现是last和lastOrDefault。

可能容易混淆，BlockingObservable也有名叫last和lastOrDefault的操作符，它们会阻塞并返回值，不是立即返回一个Observable。

7.7.1 过滤操作符

只发射最后一项数据，使用没有参数的last操作符。

示例代码

    Observable.just(1, 2, 3, 4, 5)
            .last()
            .subscribe(
                    { println("Item:$it") },
                    { println("Error:${it.message}") },
                    { println("Sequence complete") })

• Javadoc: last()

这个版本的last返回一个发射原始Observable中满足条件的最后一项数据的Observable。

    Observable.just(1, 2, 3, 4, 5)
            .last { it < 3 }
            .subscribe(
                    { println("Item:$it") },
                    { println("Error:${it.message}") },
                    { println("Sequence complete") })

• Javadoc: last(Func1)

lastOrDefault与last类似，不同的是，如果原始Observable没有发射任何值，它发射你指定的默认值。

• Javadoc: lastOrDefault(T)

这个版本的lastOrDefault可以接受一个谓词函数，如果有数据满足条件，返回的Observable就发射原始Observable满足条件的最后一项数据，否则发射默认值。

• Javadoc: lastOrDefault(T)

last和lastOrDefault默认不在任何特定的调度器上执行。

7.8 Sample

通过另一个Observable充当定时器，筛选距离定时器最近的数据进行发射。

Sample操作符定时查看一个Observable，然后发射自上次采样以来它最近发射的数据。

在某些实现中，有一个ThrottleFirst操作符的功能类似，但不是发射采样期间的最近的数据，而是发射在那段时间内的第一项数据。

RxJava将这个操作符实现为sample和throttleLast。

注意：如果自上次采样以来，原始Observable没有发射任何数据，这个操作返回的Observable在那段时间内也不会发射任何数据。

sample(别名throttleLast)的一个变体按照你参数中指定的时间间隔定时采样（TimeUnit指定时间单位）。

sample的这个变体默认在computation调度器上执行，但是你可以使用第三个参数指定其它的调度器。

    Observable.interval(50L, TimeUnit.MILLISECONDS, Schedulers.immediate())
            .limit(10)
            .sample(100L, TimeUnit.MILLISECONDS, Schedulers.newThread())
            .subscribe(
                    { println("Item:$it") },
                    { println("Error:${it.message}") },
                    { println("Sequence complete") })

• Javadoc: sample(long,TimeUnit)和throttleLast(long,TimeUnit)
• Javadoc: sample(long,TimeUnit,Scheduler)和throttleLast(long,TimeUnit,Scheduler)

sample的这个变体每当第二个Observable发射一个数据（或者当它终止）时就对原始Observable进行采样。第二个Observable通过参数传递给sample。

sample的这个变体默认不在任何特定的调度器上执行。

Observable.interval(50L, TimeUnit.MILLISECONDS, Schedulers.immediate())
            .limit(10)
            .sample(Observable.interval(100L, TimeUnit.MILLISECONDS, Schedulers.newThread()))
            .subscribe(
                    { println("Item:$it") },
                    { println("Error:${it.message}") },
                    { println("Sequence complete") })

Item:1
Item:2
Item:4
Item:6
Item:8
Item:10
Sequence complete

• Javadoc: sample(Observable)

throttleFirst与throttleLast/sample不同，在每个采样周期内，它总是发射原始Observable的第一项数据，而不是最近的一项。

throttleFirst操作符默认在computation调度器上执行，但是你可以使用第三个参数指定其它的调度器。

• Javadoc: throttleFirst(long,TimeUnit)
• Javadoc: throttleFirst(long,TimeUnit,Scheduler)

7.9 Skip

抑制Observable发射的前N项数据

使用Skip操作符，你可以忽略Observable'发射的前N项数据，只保留之后的数据。

RxJava中这个操作符叫skip。skip的这个变体默认不在任何特定的调度器上执行。

• Javadoc: skip(int)

skip的这个变体接受一个时长而不是数量参数。它会丢弃原始Observable开始的那段时间发射的数据，时长和时间单位通过参数指定。

skip的这个变体默认在computation调度器上执行，但是你可以使用第三个参数指定其它的调度器。

• Javadoc: skip(long,TimeUnit)
• Javadoc: skip(long,TimeUnit,Scheduler)

7.10 SkipLast

抑制Observable发射的后N项数据

使用SkipLast操作符修改原始Observable，你可以忽略Observable'发射的后N项数据，只保留前面的数据。

使用SkipLast操作符，你可以忽略原始Observable发射的后N项数据，只保留之前的数据。注意：这个机制是这样实现的：延迟原始Observable发射的任何数据项，直到它发射了N项数据。

skipLast的这个变体默认不在任何特定的调度器上执行。

• Javadoc: skipLast(int)

还有一个skipLast变体接受一个时长而不是数量参数。它会丢弃在原始Observable的生命周期内最后一段时间内发射的数据。时长和时间单位通过参数指定。

注意：这个机制是这样实现的：延迟原始Observable发射的任何数据项，直到自这次发射之后过了给定的时长。

skipLast的这个变体默认在computation调度器上执行，但是你可以使用第三个参数指定其它的调度器。

• Javadoc: skipLast(long,TimeUnit)
• Javadoc: skipLast(long,TimeUnit,Scheduler)

7.11 Take

只发射前面的N项数据

使用Take操作符让你可以修改Observable的行为，只返回前面的N项数据，然后发射完成通知，忽略剩余的数据。

RxJava将这个操作符实现为take函数。

如果你对一个Observable使用take(n)（或它的同义词limit(n)）操作符，而那个Observable发射的数据少于N项，那么take操作生成的Observable不会抛异常或发射onError通知，在完成前它只会发射相同的少量数据。

示例代码

    Observable.just(1, 2, 3, 4, 5, 6)
            .take(3)
            .subscribe(
                    { println("Item: $it") },
                    { println("Error: ${it.message}") },
                    { println("Sequence complete") })

Item: 1
Item: 2
Item: 3
Sequence complete

take(int)默认不任何特定的调度器上执行。

• Javadoc: take(int)

take的这个变体接受一个时长而不是数量参数。它会丢发射Observable开始的那段时间发射的数据，时长和时间单位通过参数指定。

take的这个变体默认在computation调度器上执行，但是你可以使用第三个参数指定其它的调度器。

• Javadoc: take(long,TimeUnit)
• Javadoc: take(long,TimeUnit,Scheduler)

7.12 TakeLast

发射Observable发射的最后N项数据

使用TakeLast操作符修改原始Observable，你可以只发射Observable'发射的后N项数据，忽略前面的数据。

7.12.1 taskLast.n

使用takeLast操作符，你可以只发射原始Observable发射的后N项数据，忽略之前的数据。注意：这会延迟原始Observable发射的任何数据项，直到它全部完成。

takeLast的这个变体默认不在任何特定的调度器上执行。

• Javadoc: takeLast(int)

7.12.2 takeLast.t

还有一个takeLast变体接受一个时长而不是数量参数。它会发射在原始Observable的生命周期内最后一段时间内发射的数据。时长和时间单位通过参数指定。

注意：这会延迟原始Observable发射的任何数据项，直到它全部完成。

takeLast的这个变体默认在computation调度器上执行，但是你可以使用第三个参数指定其它的调度器。

7.12.3 takeLastBuffer

还有一个操作符叫takeLastBuffer，它和takeLast类似，，唯一的不同是它把所有的数据项收集到一个List再发射，而不是依次发射一个。

• Javadoc: takeLastBuffer(int)
• Javadoc: takeLastBuffer(long,TimeUnit)
• Javadoc: takeLastBuffer(long,TimeUnit,Scheduler)
• Javadoc: takeLastBuffer(int,long,TimeUnit)
• Javadoc: takeLastBuffer(int,long,TimeUnit,Scheduler)