RX操作符之过滤操作一（debounce、filter、ofType、takeLast、last、lastOrDefault、takeLastBuffer、skip、skipLast、take）

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

Debounce

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

注：这个操作符会会接着最后一项数据发射原始Observable的

onCompleted

通知，即使这个通知发生在你指定的时间窗口内（从最后一项数据的发射算起）。也就是说，

onCompleted

通知不会触发限流。

Integer[]items = {1,2,3,4,5,6,7,8};

Observable myObservable = Observable.from(items)
.debounce(2,TimeUnit.SECONDS);

Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



二、filter
只发射符合条件的数据，

Filter

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

Integer[]items = {1,2,3,4,5,6,7,8};

Observable myObservable = Observable.from(items)
.filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer < 7;
}
});

Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



三、ofType

ofType只返回指定类型的数据

Object [] items = {"dd",1,"ff",33,"fr",6,8,"zga",5};
Observable myObservable = Observable.from(items)
.ofType(Integer.class);
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



四、takeLast
发射Observable发射的最后N项数据，

TakeLast

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

Integer[]items = {1,2,3,4,5,6,7,8};
Observable myObservable = Observable.from(items)
.takeLast(5);
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



五、last

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

注：

Last

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

TakeLast(1)

。

Integer[]items = {1,2,3,4,5,6,7,8};
Observable myObservable = Observable.from(items)
.last();
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



六、lastOrDefault

lastOrDefault

与

last

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

Integer[]items = {};
Observable myObservable = Observable.from(items)
.lastOrDefault(3);
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



七、takeLastBuffer

takeLastBuffer

和

takeLast

类似，唯一的不同是它把所有的数据项收集到一个

List

再发射，而不是依次发射一个。

Integer[]items = {1,2,3,4,5,6,7,8};
Observable myObservable = Observable.from(items)
.takeLastBuffer(4);
Subscriber<List<Integer>> mySubscriber =new Subscriber<List<Integer>>() {
@Override
public void onNext(List<Integer> list) {
System.out.println("onNext方法中集合的長度是................."+list.size());
for(int i=0;i<list.size();i++){
System.out.println("onNext循環................."+list.get(i));
}
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



八、skip
跳过之前的n项，从n+1项开始发射数据

Integer[]items = {1,2,3,4,5,6,7,8};
Observable myObservable = Observable.from(items)
.skip(4);
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



九、skipLast

SkipLast

操作符，你可以忽略原始Observable发射的后N项数据，只保留之前的数据。注意：这个机制是这样实现的：延迟原始Observable发射的任何数据项，直到它发射了N项数据。
Integer[]items = {1,2,3,4,5,6,7,8};
Observable myObservable = Observable.from(items)
.skipLast(4);
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：



十、take

Take

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

Integer[]items = {1,2,3,4,5,6,7,8};
Observable myObservable = Observable.from(items)
.take(4);
Subscriber<Integer> mySubscriber =new Subscriber<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println("onNext................."+integer);
}

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

@Override
public void onError(Throwable e) {
System.out.println("onError....................");
}
};

myObservable.subscribe(mySubscriber);

运行结果：