过滤操作

这个页面展示的操作符可用于过滤和选择Observable发射的数据序列。

Debounce

仅在过了一段指定的时间还没发射数据时才发射一个数据

Debounce操作符会过滤掉发射速率过快的数据项。

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

注意：这个操作符会会接着最后一项数据发射原始Observable的onCompleted通知，即使这个通知发生在你指定的时间窗口内（从最后一项数据的发射算起）。也就是说，onCompleted通知不会触发限流。

throttleWithTimeout

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

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

debounce

debounce操作符的一个变体通过对原始Observable的每一项应用一个函数进行限流，这个函数返回一个Observable。如果原始Observable在这个新生成的Observable终止之前发射了另一个数据，debounce会抑制(suppress)这个数据项。

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

Distinct

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

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

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

distinct()

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

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
Observable.just(1, 2, 1, 1, 2, 3)
          .distinct()
          .subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出

1
2
3
4
Next: 1
Next: 2
Next: 3
Sequence complete.

distinct(Func1)

这个操作符有一个变体接受一个函数。这个函数根据原始Observable发射的数据项产生一个Key，然后，比较这些Key而不是数据本身，来判定两个数据是否是不同的。

distinctUntilChanged

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

distinctUntilChanged(Func1)

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

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

ElementAt

只发射第N项数据

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

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

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

elementAtOrDefault

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

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

Filter

只发射通过了谓词测试的数据项

Filter操作符使用你指定的一个谓词函数测试数据项，只有通过测试的数据才会被发射。

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

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
Observable.just(1, 2, 3, 4, 5)
          .filter(new Func1<Integer, Boolean>() {
              @Override
              public Boolean call(Integer item) {
                return( item < 4 );
              }
          }).subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出

1
2
3
4
Next: 1
Next: 2
Next: 3
Sequence complete.

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

ofType

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

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

First

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

如果你只对Observable发射的第一项数据，或者满足某个条件的第一项数据感兴趣，你可以使用First操作符。

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

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

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

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

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

IgnoreElements

不发射任何数据，只发射Observable的终止通知

IgnoreElements操作符抑制原始Observable发射的所有数据，只允许它的终止通知（onError或onCompleted）通过。

如果你不关心一个Observable发射的数据，但是希望在它完成时或遇到错误终止时收到通知，你可以对Observable使用ignoreElements操作符，它会确保永远不会调用观察者的onNext()方法。

RxJava将这个操作符实现为ignoreElements。

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

Last

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

如果你只对Observable发射的最后一项数据，或者满足某个条件的最后一项数据感兴趣，你可以使用Last操作符。

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

在RxJava中的实现是last和lastOrDefault。

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

过滤操作符

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

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
Observable.just(1, 2, 3)
          .last()
          .subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出

1
2
Next: 3
Sequence complete.

这个版本的last也是接受一个谓词函数，返回一个发射原始Observable中满足条件的最后一项数据的Observable。

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

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

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

Sample

定期发射Observable最近发射的数据项

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

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

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

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

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

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

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

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

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

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

Skip

抑制Observable发射的前N项数据

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

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

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

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

SkipLast

抑制Observable发射的后N项数据

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

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

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

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

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

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

Take

只发射前面的N项数据

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

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

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

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
          .take(4)
          .subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出

1
2
3
4
5
Next: 1
Next: 2
Next: 3
Next: 4
Sequence complete.

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

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

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

TakeLast

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

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

taskLast.n

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

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

takeLast.t

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

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

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

takeLastBuffer

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