RxJava 1.x 笔记：过滤型操作符

秒速五厘米 2022-06-12 04:50 207阅读 0赞

我真的是奇怪，上下班的路上看书、看文章学习的劲头特别大，到了周末有大把的学习时间，反而不珍惜，总想打游戏，睡前才踏踏实实地写了篇文章，真是服了自己！

> 本文内容为 [RxJava 官方文档][RxJava] 学习笔记  
> 作者：[shixinzhang][]

读完本文你将了解：

*  过滤型操作符
    
     *  Debounce
        
         *  debounce
         *  throttleWithTimeout
     *  Distinct
        
         *  distinct
         *  distinctkeySelector
         *  distinctUntilChanged
         *  distinctUntilChangedkeySelector
     *  ElementAt
        
         *  elementAt
         *  elementAtOrDefault
     *  Filter
        
         *  filter
         *  ofType
     *  First
        
         *  first
         *  firstpredicate
         *  firstOrDefault
         *  firstOrDefaultpredicate
         *  takeFirst
         *  single
     *  Last
     *  Take
        
         *  takecount
         *  taketime unit
     *  TakeLast
     *  IgnoreElements
     *  Sample
        
         *  sample
         *  throttleFirst
     *  Skip
        
         *  skipcount
         *  skiptime unit
     *  SkipLast
 *  代码地址
 *  Thanks

# 过滤型操作符 #

过滤型操作符即对 Observable 的数据进行过滤，选择性地发射出去。

## Debounce ##

`Debounce` 的作用是：控制发射速率。

每产生一个数据后，如果在规定的间隔时间内没有别的数据产生，就会发射这个数据，否则忽略该数据。

![这里写图片描述][SouthEast]

RxJava 的实现有两种：`debounce` 和 `throttleWithTimeout`。

### debounce ###

![这里写图片描述][SouthEast 1]

`debounce` 接收两个参数，第一个指定过滤的时间间隔，第二个参数指定单位。

public final Observable<T> debounce(long timeout, TimeUnit unit) {
        return debounce(timeout, unit, Schedulers.computation());
    }
    public final Observable<T> debounce(long timeout, TimeUnit unit, Scheduler scheduler) {
        return lift(new OperatorDebounceWithTime<T>(timeout, unit, scheduler));
    }

使用例子：

private void filteringWithDebounce() {
        Observable
                .unsafeCreate(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            SystemClock.sleep(i % 5 * 1000);
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .subscribeOn(Schedulers.computation())
                .debounce(2, TimeUnit.SECONDS)
                .subscribe(this.<Integer>getPrintSubscriber());
    
    }

在上面的例子中，我们发射 0 到 9 共 10 个数据，每个延迟 i % 5 秒，也就是延迟 0 到 4 秒，过滤时间为 2 秒，所以最终发射结果：

![这里写图片描述][SouthEast 2]

### throttleWithTimeout ###

![这里写图片描述][SouthEast 3]

`throttleWithTimeout` 和 `debounce` 作用一样，通过源码可以看到，它也是调用的 `debounce`:

public final Observable<T> throttleWithTimeout(long timeout, TimeUnit unit) {
        return debounce(timeout, unit);
    }

例子就不演示了。

## Distinct ##

![这里写图片描述][SouthEast 4]

`Distinct` 即“去重”，很好理解。

RxJava 中的实现有 4 种。

### distinct() ###

![这里写图片描述][SouthEast 5]

第一种就是最简单的无参数 `distinct()`:

public final Observable<T> distinct() {
        return lift(OperatorDistinct.<T> instance());
    }

使用也很简单：

private void filteringWithDistinct() {
        Observable.from(Arrays.asList(1,3,1,3,4))
                .distinct()
                .subscribe(this.<Integer>getPrintSubscriber());
    }

结果就和我们期望的一样：

![这里写图片描述][SouthEast 6]

### distinct(keySelector) ###

![这里写图片描述][SouthEast 7]

第二种，`distinct(keySelector)` 允许我们传入一个函数作为参数，这个函数返回了决定是否重复的 Key。

public final <U> Observable<T> distinct(Func1<? super T, ? extends U> keySelector) {
        return lift(new OperatorDistinct<T, U>(keySelector));
    }

使用例子：

private void filteringWithDistinct2() {
    
        Observable.from(Arrays.asList(1,3,1,3,4))
                .distinct(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer item) {
                        return item % 2;
                    }
                })
                .subscribe(this.<Integer>getPrintSubscriber());
    }

在函数中我们以 item % 2 的结果作为判断是否重复的依据，源 Observable 发射的数据中，对二求余的结果只有 1 和 0，因此输出结果为：

![这里写图片描述][SouthEast 8]

### distinctUntilChanged() ###

![这里写图片描述][SouthEast 9]

`distinctUntilChanged()` 也是去重，不过每个元素只跟前面一个元素比较，如果和前面的一样就去除，否则就发射，不会和其他位置的比较。

public final Observable<T> distinctUntilChanged() {
        return lift(OperatorDistinctUntilChanged.<T> instance());
    }

使用例子：

private void filteringWithDistinctUntilChanged() {
        Observable.from(Arrays.asList(1,1,3,1,3,4,4))
                .distinctUntilChanged()
                .subscribe(this.<Integer>getPrintSubscriber());
    }

运行结果：

![这里写图片描述][SouthEast 10]

可以看到，输出的结果还是有重复，去掉的是和前一个元素重复的元素。

### distinctUntilChanged(keySelector) ###

![这里写图片描述][SouthEast 11]

`distinctUntilChanged(keySelector)` 就是 `distinct(keySelector)` 和 `distinctUntilChanged()` 的结合体，传入的参数决定是否重复，并且只和前一个元素比较。

就不写例子了。

## ElementAt ##

![这里写图片描述][SouthEast 12]

`ElementAt` 和名字一样，只发射指定位置的元素（从 0 开始）。

RxJava 对应的实现有 2 种。

### elementAt ###

![这里写图片描述][SouthEast 13]

`elementAt` 和规范一样，只发射指定位置的元素，

public final Observable<T> elementAt(int index) {
        return lift(new OperatorElementAt<T>(index));
    }
    private OperatorElementAt(int index, T defaultValue, boolean hasDefault) {
        if (index < 0) {
            throw new IndexOutOfBoundsException(index + " is out of bounds");
        }
        this.index = index;
        this.defaultValue = defaultValue;
        this.hasDefault = hasDefault;
    }

从 `OperatorElementAt` 的构造函数我们可以看到当指定的位置小于 0 时，会抛出 `java.lang.IndexOutOfBoundsException` 异常。  
当 index > 数据总数时，会发射　`onError` 事件：

private void filteringWithElementAt() {
        Observable.range(0, 10)
                .elementAt(12)
                .subscribe(this.<Integer>getPrintSubscriber());
    }

运行结果：

07-15 22:00:52.425 11596-11596/top.shixinzhang.rxjavademo I/System.out: onError: 12 is out of bounds

### elementAtOrDefault ###

![这里写图片描述][SouthEast 14]

`elementAtOrDefault` 和 `elementAt` 的区别在于，当指定的 index 大于源 Observable 发射的数据长度时，不会发射 onError 事件，而是发射预备的默认值；不过 index < 0 时还是会抛出异常。

public final Observable<T> elementAtOrDefault(int index, T defaultValue) {
        return lift(new OperatorElementAt<T>(index, defaultValue));
    }
    private OperatorElementAt(int index, T defaultValue, boolean hasDefault) {
        if (index < 0) {
            throw new IndexOutOfBoundsException(index + " is out of bounds");
        }
        this.index = index;
        this.defaultValue = defaultValue;
        this.hasDefault = hasDefault;
    }

使用例子：

private void filteringWithElementAtDefault() {
        Observable.range(0, 10)
                .elementAtOrDefault(12, 222)
                .subscribe(this.<Integer>getPrintSubscriber());
    
    }

运行结果：

07-15 22:06:27.882 16870-16870/top.shixinzhang.rxjavademo I/System.out: onNext: 222
    07-15 22:06:27.882 16870-16870/top.shixinzhang.rxjavademo I/System.out: onCompleted

## Filter ##

![这里写图片描述][SouthEast 15]

`Filter` 只发射符合要求的元素。

RxJava 中的实现有 2 种。

### filter ###

![这里写图片描述][SouthEast 16]

`filter(predicate)` 的参数指定了要发射的元素需要满足的条件，不满足就不会发射。

public final Observable<T> filter(Func1<? super T, Boolean> predicate) {
        return unsafeCreate(new OnSubscribeFilter<T>(this, predicate));
    }

使用例子：

private void filteringWithFilter() {
        Observable.range(0, 10)
                .filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer item) {
                        return item > 5;
                    }
                })
                .subscribe(this.<Integer>getPrintSubscriber());
    }

例子中，我们只允许大于 5 的数据发射。运行结果：

07-15 22:09:38.773 19765-19765/top.shixinzhang.rxjavademo I/System.out: onNext: 6
    07-15 22:09:38.773 19765-19765/top.shixinzhang.rxjavademo I/System.out: onNext: 7
    07-15 22:09:38.773 19765-19765/top.shixinzhang.rxjavademo I/System.out: onNext: 8
    07-15 22:09:38.773 19765-19765/top.shixinzhang.rxjavademo I/System.out: onNext: 9
    07-15 22:09:38.773 19765-19765/top.shixinzhang.rxjavademo I/System.out: onCompleted

### ofType ###

![这里写图片描述][SouthEast 17]

`ofType(klass)` 的参数指定符合要求的数据类型，最终还是调用的 `filter`：

public final <R> Observable<R> ofType(final Class<R> klass) {
        return filter(InternalObservableUtils.isInstanceOf(klass)).cast(klass);
    }

使用例子：

private void filteringWithOfType() {
        Observable.range(0, 10)
                .ofType(String.class)
                .subscribe(this.<String>getPrintSubscriber());
    }

可以看到 `ofType` 返回的 Observable 的数据类型就是参数的类型。运行结果：

07-15 22:14:38.979 24199-24199/top.shixinzhang.rxjavademo I/System.out: onCompleted

## First ##

![这里写图片描述][SouthEast 18]

`First` 的作用的就是只发射第一个元素（或者是第一个满足要求的元素）。

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

### first ###

![这里写图片描述][SouthEast 19]

`first()` 只发射第一个数据，如果源 Observable是空的话，会抛出 `NoSuchElementException` 异常。

public final Observable<T> first() {
        return take(1).single();
    }

可以看到它调用的是 `take(1).single()`，这两个操作符我们这篇文章后面介绍。

运行结果很简单，这里就暂不演示了。

### first(predicate) ###

![这里写图片描述][SouthEast 20]

`first(predicate)` 只发射符合要求的第一个元素。

public final Observable<T> first(Func1<? super T, Boolean> predicate) {
        return takeFirst(predicate).single();
    }

使用例子：

private void filteringWithFirst() {
        Observable.range(4, 10)
                .first(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer > 5;
                    }
                })
                .subscribe(this.<Integer>getPrintSubscriber());
    }

我们只发射第一个大于 5 的数字，也就是 6 喽。运行结果：

07-15 22:23:02.228 31262-31262/top.shixinzhang.rxjavademo I/System.out: onNext: 6
    07-15 22:23:02.228 31262-31262/top.shixinzhang.rxjavademo I/System.out: onCompleted

### firstOrDefault ###

![这里写图片描述][SouthEast 21]

`firstOrDefault` 和名字一样，如果没有第一个元素就发射默认的。

public final Observable<T> firstOrDefault(T defaultValue) {
        return take(1).singleOrDefault(defaultValue);
    }

使用例子：

private void filteringWithFirstOrDefault() {
        Observable.empty()
                .firstOrDefault(33)
                .subscribe(getPrintSubscriber());
    }

运行结果：

07-15 22:26:28.375 2910-2910/top.shixinzhang.rxjavademo I/System.out: onNext: 33
    07-15 22:26:28.375 2910-2910/top.shixinzhang.rxjavademo I/System.out: onCompleted

### firstOrDefault(predicate) ###

![这里写图片描述][SouthEast 22]

`firstOrDefault(predicate)` 返回第一个符合要求的，没有的话就返回默认的，也很好理解。

public final Observable<T> firstOrDefault(T defaultValue, Func1<? super T, Boolean> predicate) {
        return takeFirst(predicate).singleOrDefault(defaultValue);
    }

使用例子：

private void filteringWithFirstOrDefault() {
        Observable.range(0 , 4)
                .firstOrDefault(33, new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer > 5;
                    }
                })
                .subscribe(getPrintSubscriber());
    }

运行结果：

07-15 22:26:28.375 2910-2910/top.shixinzhang.rxjavademo I/System.out: onNext: 33
    07-15 22:26:28.375 2910-2910/top.shixinzhang.rxjavademo I/System.out: onCompleted

### takeFirst ###

![这里写图片描述][SouthEast 23]

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

也就是说 `takeFirst` 比 `first` 温柔一点，不会抛异常。

### single ###

![这里写图片描述][SouthEast 24]

`single` 也是只发射一个数据，但是如果源 Observable 发射多个数据，就会发射 `onError` 事件：

07-15 22:37:15.774 12609-12609/top.shixinzhang.rxjavademo I/System.out: onError: Sequence contains too many elements

如果源 Observable 没有数据，也会发射 `onError` 事件：

07-15 22:38:30.700 13779-13779/top.shixinzhang.rxjavademo I/System.out: onError: Sequence contains no elements

**也就是说，`single` 是用来检验并获取只有一个元素的 Observable 发射的数据。**

> single 也有传递符合要求函数、默认值的变体，这里就暂不赘述了。

## Last ##

有 First 当然就有 Last。

![这里写图片描述][SouthEast 25]

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

Last 的变体和 First 差不多，这里就不赘述了。

## Take ##

![这里写图片描述][SouthEast 26]

`first` 很多都是使用 `take` 实现的。

`Take` 操作符的作用是：**只保留前面的若干项数据**。

RxJava 对应的实现为 `take`。

### take(count) ###

![这里写图片描述][SouthEast 27]

`take(count)` 的参数指定要保留的数据项。

public final Observable<T> take(final int count) {
        return lift(new OperatorTake<T>(count));
    }

另外 `limit` 也只是 `take` 的别名：

public final Observable<T> limit(int count) {
        return take(count);
    }

使用例子：

private void filteringWithTake() {
        Observable.range(0 , 10)
                .limit(3)
                .subscribe(this.<Integer>getPrintSubscriber());
    }

运行结果：

07-15 22:51:49.895 25751-25751/top.shixinzhang.rxjavademo I/System.out: onNext: 0
    07-15 22:51:49.895 25751-25751/top.shixinzhang.rxjavademo I/System.out: onNext: 1
    07-15 22:51:49.895 25751-25751/top.shixinzhang.rxjavademo I/System.out: onNext: 2
    07-15 22:51:49.895 25751-25751/top.shixinzhang.rxjavademo I/System.out: onCompleted

### take(time, unit) ###

![这里写图片描述][SouthEast 28]

另一种变体是 `take(long time, TimeUnit unit)`，参数为时间，即只发射指定时间之内（小于该时间）发射的数据，超时的数据都不会发射。

public final Observable<T> take(long time, TimeUnit unit) {
        return take(time, unit, Schedulers.computation());
    }
    public final Observable<T> take(long time, TimeUnit unit, Scheduler scheduler) {
        return lift(new OperatorTakeTimed<T>(time, unit, scheduler));
    }

使用例子：

private void filteringWithTake() {
        Observable
                .unsafeCreate(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            SystemClock.sleep(1_000);
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .subscribeOn(Schedulers.computation())
                .take(3, TimeUnit.SECONDS)
                .subscribe(this.<Integer>getPrintSubscriber());
    }

每隔一秒发射一个数据，然后指定只要 3 秒内发射的数据。运行结果：

07-15 22:56:01.304 29363-29388/top.shixinzhang.rxjavademo I/System.out: onNext: 0
    07-15 22:56:02.306 29363-29388/top.shixinzhang.rxjavademo I/System.out: onNext: 1
    07-15 22:56:03.302 29363-29387/top.shixinzhang.rxjavademo I/System.out: onCompleted

## TakeLast ##

![这里写图片描述][SouthEast 29]

有从前开始拿，自然就有从后开始拿，`TakeLast` 就是这个作用。

使用 `TakeLast` 操作符可以只发射 Observable 发射的后 N 项数据，忽略前面的数据。

RxJava 中的实现有 `takeLast`，它的参数可以是个数也可以是时间。

还有一种实现是 `takeLastBuffer`：

![这里写图片描述][SouthEast 30]

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

## IgnoreElements ##

![这里写图片描述][SouthEast 31]

`IgnoreElements` 的作用是不发射任何数据，只发射结束事件（ onError or onCompleted）。

当你不在乎发射的内容，只希望在它完成时或遇到错误终止时收到通知，可以使用这个操作符。

> `ignoreElements` 会确保永远不会调用观察者的onNext()方法

RxJava 的实现是 `ignoreElements`：

public final Observable<T> ignoreElements() {
        return lift(OperatorIgnoreElements.<T> instance());
    }
    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> child) {
        Subscriber<T> parent = new Subscriber<T>() {
    
            @Override
            public void onCompleted() {
                child.onCompleted();
            }
    
            @Override
            public void onError(Throwable e) {
                child.onError(e);
            }
    
            @Override
            public void onNext(T t) {
                // ignore element
            }
    
        };
        child.add(parent);
        return parent;
    }

可以看到，它的 `onNext()` 方法没有传递事件。

## Sample ##

`Sample` 的作用是：定时发射 Observable 最新发射的数据。

![这里写图片描述][SouthEast 32]

`Sample` 操作符会周期性地查看源 Observable，发射自出上次查看以来，最新发射的数据。

RxJava 中有三种实现：`sample`, `throttleFirst`, `throttleLast`。

### sample ###

![这里写图片描述][SouthEast 33]

`sample()` 的参数指定定期查看的时间间隔：

public final Observable<T> sample(long period, TimeUnit unit) {
        return sample(period, unit, Schedulers.computation());
    }
    public final Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler) {
        return lift(new OperatorSampleWithTime<T>(period, unit, scheduler));
    }

`throttleLast` 和 `sample` 一样，只不过名称不同：

public final Observable<T> throttleLast(long intervalDuration, TimeUnit unit) {
        return sample(intervalDuration, unit);
    }

使用例子：

private void filteringWithSample() {
        Observable
                .unsafeCreate(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            SystemClock.sleep( i % 5 * 1000);
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .subscribeOn(Schedulers.computation())
                .sample(3, TimeUnit.SECONDS)
                .subscribe(this.<Integer>getPrintSubscriber());
    }

例子中，我们每隔 i % 5 秒发射 10 个数据，然后每隔 3 秒去查看一次，发射距离上次查看，最新发射的元素。运行结果：

07-15 23:12:55.915 12800-12815/top.shixinzhang.rxjavademo I/System.out: onNext: 0
    07-15 23:12:58.915 12800-12815/top.shixinzhang.rxjavademo I/System.out: onNext: 2
    07-15 23:13:01.916 12800-12815/top.shixinzhang.rxjavademo I/System.out: onNext: 3
    07-15 23:13:07.916 12800-12815/top.shixinzhang.rxjavademo I/System.out: onNext: 6
    07-15 23:13:10.916 12800-12815/top.shixinzhang.rxjavademo I/System.out: onNext: 7
    07-15 23:13:13.916 12800-12815/top.shixinzhang.rxjavademo I/System.out: onNext: 8
    07-15 23:13:15.499 12800-12816/top.shixinzhang.rxjavademo I/System.out: onNext: 9
    07-15 23:13:15.499 12800-12816/top.shixinzhang.rxjavademo I/System.out: onCompleted

### throttleFirst ###

![这里写图片描述][SouthEast 34]

`throttleFirst` 也是隔一段时间去查看一次，不同的是它发射的是这段时间里第一个发射的数据，而不是最新的。

public final Observable<T> throttleFirst(long windowDuration, TimeUnit unit) {
        return throttleFirst(windowDuration, unit, Schedulers.computation());
    }
    public final Observable<T> throttleFirst(long skipDuration, TimeUnit unit, Scheduler scheduler) {
        return lift(new OperatorThrottleFirst<T>(skipDuration, unit, scheduler));
    }

例子：

private void filteringWithThrottleFirst() {
        Observable
                .unsafeCreate(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            SystemClock.sleep( i % 5 * 1000);
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .subscribeOn(Schedulers.computation())
                .throttleFirst(3, TimeUnit.SECONDS)
                .subscribe(this.<Integer>getPrintSubscriber());
    }

运行结果：

07-15 23:18:00.798 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 0
    07-15 23:18:03.813 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 2
    07-15 23:18:06.815 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 3
    07-15 23:18:10.816 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 4
    07-15 23:18:13.818 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 7
    07-15 23:18:16.820 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 8
    07-15 23:18:20.822 17008-17648/top.shixinzhang.rxjavademo I/System.out: onNext: 9
    07-15 23:18:20.822 17008-17648/top.shixinzhang.rxjavademo I/System.out: onCompleted

## Skip ##

`Skip` 的作用是，跳过指定数量的数据，发射后面的数据。

![这里写图片描述][SouthEast 35]

RxJava 中的实现有两种，都叫 `skip`，不同的是一个是按个数算，一个是按时间算。

### skip(count) ###

![这里写图片描述][SouthEast 36]

public final Observable<T> skip(int count) {
        return lift(new OperatorSkip<T>(count));
    }

使用例子：

private void filteringWithSkip() {
        Observable.range(0 , 10)
                .skip(3)
                .subscribe(this.<Integer>getPrintSubscriber());
    }

运行结果：

07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 3
    07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 4
    07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 5
    07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 6
    07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 7
    07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 8
    07-15 23:22:14.472 21075-21075/top.shixinzhang.rxjavademo I/System.out: onNext: 9
    07-15 23:22:14.473 21075-21075/top.shixinzhang.rxjavademo I/System.out: onCompleted

### skip(time, unit) ###

![这里写图片描述][SouthEast 37]

`skip(time, unit)` 的参数指定要跳过前指定时间内发射的数据。

public final Observable<T> skip(long time, TimeUnit unit) {
        return skip(time, unit, Schedulers.computation());
    }
    public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler) {
        return unsafeCreate(new OnSubscribeSkipTimed<T>(this, time, unit, scheduler));
    }

## SkipLast ##

既然有跳过从头开始的数据，自然也有跳过从后开始的数据，这就是 `SkipLast` 的作用。

![这里写图片描述][SouthEast 38]

RxJava 中的实现也有两种，按时间和按个数，这里就暂不赘述了。

# [代码地址][Link 1] #

# Thanks #

[http://reactivex.io/documentation/operators.html][RxJava]  
[https://github.com/mcxiaoke/RxDocs/blob/master/Operators.md][https_github.com_mcxiaoke_RxDocs_blob_master_Operators.md]  
[http://blog.csdn.net/job\_hesc/article/details/46495281][http_blog.csdn.net_job_hesc_article_details_46495281]

[RxJava]: http://reactivex.io/documentation/operators.html
[shixinzhang]: http://blog.csdn.net/u011240877
[SouthEast]: /images/20220612/d7e4461a5d7948d7b2abc24d3bca8597.png
[SouthEast 1]: /images/20220612/afbd754afbc443c68d7f32d98e51cbc8.png
[SouthEast 2]: /images/20220612/a8031c0563df494bb1c03f20c9ac571d.png
[SouthEast 3]: /images/20220612/3516c65465a14d498b75fa61a2fe7d6a.png
[SouthEast 4]: /images/20220612/305560d0691f419fa2a349b4ffc7eb64.png
[SouthEast 5]: /images/20220612/811689b4200a4003bfc6078f219b7e0c.png
[SouthEast 6]: /images/20220612/4b9744c30ace452390fd47813575ef5b.png
[SouthEast 7]: /images/20220612/9612b9c41fc243599648db732d03a684.png
[SouthEast 8]: /images/20220612/66316a41238b46eb8a6214349c3a5b45.png
[SouthEast 9]: /images/20220612/e45a960f259a4ee8b536ec3687b98ac2.png
[SouthEast 10]: /images/20220612/5759bc09c5c441c79cb900c8e44ee7ef.png
[SouthEast 11]: /images/20220612/52bc0895653d4cc081c520450b69393c.png
[SouthEast 12]: /images/20220612/18eb9eabc18c4a70ab4a70adbc8b6341.png
[SouthEast 13]: /images/20220612/6f6e34c092734ce3a3a7ae8f87f2712f.png
[SouthEast 14]: /images/20220612/797118fe7344415ab2223382a94205b2.png
[SouthEast 15]: /images/20220612/605a90cd15fa41569f03203b32620c53.png
[SouthEast 16]: /images/20220612/078ef84da25f429aa96893dbd4b1126f.png
[SouthEast 17]: /images/20220612/eb95369111154b5a9909eb585e2bfee3.png
[SouthEast 18]: /images/20220612/d8f19901cfaf4cc897e7cf2efc3984d0.png
[SouthEast 19]: /images/20220612/402d039306d1421897283631243b9fae.png
[SouthEast 20]: /images/20220612/f14ed4229ecc4e2c979606503eef606a.png
[SouthEast 21]: /images/20220612/629648dbe33844b98f0fabf3ce530479.png
[SouthEast 22]: /images/20220612/6f2f5448b3ca41c5984764a54c991cee.png
[SouthEast 23]: /images/20220612/f1055d1571554792b80fb1b87215ed0f.png
[SouthEast 24]: /images/20220612/e2140ab9339c42528f2524efc840546a.png
[SouthEast 25]: /images/20220612/5191fe80c6f94548a49a175f85eb17a2.png
[SouthEast 26]: /images/20220612/a5e1c2fbdcf44cf287938ea18f370e77.png
[SouthEast 27]: /images/20220612/58ada66c8c464e458eaa43d7e4931e63.png
[SouthEast 28]: /images/20220612/9f4d44d5eb184dc388d1604975ebed1d.png
[SouthEast 29]: /images/20220612/755a358d1d47490ea991f3b6f169a22f.png
[SouthEast 30]: /images/20220612/a4e4aa1d4d774964b153324b1ab1177f.png
[SouthEast 31]: /images/20220612/517cce5e257443c0ab0348ff2c965dfe.png
[SouthEast 32]: /images/20220612/fce5fc4c2a394f789c0205c2a6efbd9f.png
[SouthEast 33]: /images/20220612/0630eb9048d24dfcb39faa94934dadd2.png
[SouthEast 34]: /images/20220612/887ec2157a844ae390a8dfcb0f2137d8.png
[SouthEast 35]: /images/20220612/8ea014a1eeac4f3dbc6ee0ae2f26577d.png
[SouthEast 36]: /images/20220612/fd059261fd564d368fa6764f72581bc3.png
[SouthEast 37]: /images/20220612/967a19a4901640ecb4994a1935ce8e16.png
[SouthEast 38]: /images/20220612/75b8828350604b0fa83a14bfea3aeb63.png
[Link 1]: https://github.com/shixinzhang/LearnAndroid/tree/dev/rxjavademo/src/main/java/top/shixinzhang/rxjavademo
[https_github.com_mcxiaoke_RxDocs_blob_master_Operators.md]: https://github.com/mcxiaoke/RxDocs/blob/master/Operators.md
[http_blog.csdn.net_job_hesc_article_details_46495281]: http://blog.csdn.net/job_hesc/article/details/46495281