此篇内容均是来自书籍《RxJava响应式编程》李衍顺 著
3.3 过滤操作符
过滤操作符的作用类似于SQL里的where语句，能够让Observable只发送出满足条件的数据。
3.3.1 debounce
debounce操作符是用来限流的，debounce过滤掉的数据会被丢弃掉。这个操作符又被分为throttleWithTimeout和debounce两个操作符。
throttleWithTimeout这个操作符通过时间来限流，每次接收到源Observable发送出来的数据后会进行计时，如果在设定好的时间结束前又接收到源Observable发送的新数据，那么上一个数据就会被丢弃，同时重新开始计时。如果每次都是在计时结束前接收到新数据，那么这个限流就会走向极端：只发送最后一个数据
下面我们来创建一个Observable，会每隔100毫秒发送一个数据，当要发送的数据是3的倍数时，下一个数据就延迟到300毫秒后发送。我们使用throttleWithTimeout来过滤一下这个Observable，设定过滤时间为200毫秒，也就是说发送间隔小于200毫秒的数据会被过滤掉。
private void debounce(){
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for(int i=0; i<10; i++){
if(!subscriber.isUnsubscribed()){
subscriber.onNext(i);
}
int sleep = 100;
if(i%3==0){
sleep = 300;
}
try {
Thread.sleep(sleep);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
subscriber.onCompleted();
}
}).subscribeOn(Schedulers.computation())
.throttleWithTimeout(200, TimeUnit.MILLISECONDS)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("throttleWithTimeout: " + integer);
}
});
Observable.interval(1000, TimeUnit.MILLISECONDS)
.debounce(new Func1<Long, Observable<Long>>() {
@Override
public Observable<Long> call(Long aLong) {
return Observable.timer(aLong%2*1500, TimeUnit.MILLISECONDS);
}
}).subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
System.out.println("debounce: " + aLong);
}
});
}
但是发现结果什么都没有输出，不知道是哪里写错啦，需要进一步研究。
3.3.2 distinct
distinct操作符的用处就是去重，非常好理解。所有重复数据都会被过滤掉，其作用范围是全局的，也就是说所有发送出去的数据肯定是没有重复的。还要一个特殊的实现操作符：distinctUntilChanged，这个操作符被用来过滤掉连续的重复数据，作用范围是局部的，也就是说发送出来的数据可能会有重复，但是这些重复的数据肯定不是连续的。
private void distinct() {
Observable.just(1,2,3,4,5,4,3,2,1)
.distinct()
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("distinct: "+ integer);
}
});
Observable.just(1,2,3,3,3,1,2)
.distinctUntilChanged()
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("untilchanged: "+ integer);
}
});
}
distinct: 1
distinct: 2
distinct: 3
distinct: 4
distinct: 5
untilchanged: 1
untilchanged: 2
untilchanged: 3
untilchanged: 1
untilchanged: 2
3.3.3 elementAt
elementAt只会过滤出源Observable发送出来的顺序为N的数据。同数组的下标一样，这里的N也是从0开始的，也就是说如果N为0，将只会过滤出源Observable发送的第一个数据。
private void elementAt(){
Observable.just(2,4,4,2,5)
.elementAt(2)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("elementAt: "+ integer);
}
});
}
结果：
elementAt: 4
3.3.4 filter
filter操作符根据一个函数来进行过滤操作。源Observable发送的数据作为参数传递到函数里，如果函数的返回值为true，则这个数据将继续被发送出去，否则数据将会被过滤掉。
private void filter(){
Observable.just(0,1,2,3,4,5)
.filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer<3;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("Filter: "+ integer);
}
});
}
结果为：
Filter: 0
Filter: 1
Filter: 2
3.3.5 first 和 last
first操作符只会返回第一条或者满足条件的第一条数据。last操作符只会返回最后一条或者满足条件的最后一条数据。
另外还有一个特殊的Observable：BlockingObservable， 可以使用Observable.toBlocking或者BlockingObservable.from方法来将一个Observable对象转化为BlockingObservable对象。BlockingObservable不会对Observable做任何处理，只会阻塞住。只有当满足条件的数据发送出来的时候才会发送第一个BlockingObservable对象。BlockingObservable也有可以配合使用的first操作符，不过其first操作符返回的不是Observable，而是对应的数据。
private void first_last(){
Observable.just(1,2,3,4,5)
.first(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer > 3;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("First: " + integer);
}
});
int result = Observable.just(3,4,5,6,7)
.toBlocking()
.first(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer > 4;
}
});
System.out.println("blocking: "+ result);
}
结果
First: 4
blocking: 5
3.3.6 skip和take， skipLast 和takeLast
skip操作符能够将源Observable发送的数据过滤掉前n项，与之相反，take操作符则只取前n项。另外还有skipLast和takeLast，分别是从后面进行相应的过滤操作。
private void skip_take(){
Observable.just(1,2,3,4,5)
.skip(2)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("skip: "+ integer);
}
});
Observable.just(2,3,4,5,6)
.take(3)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("take: "+ integer);
}
});
}
结果：
skip: 3
skip: 4
skip: 5
take: 2
take: 3
take: 4
3.3.7 sample 和 throttleFirst
sample操作符会指定一段时长，在每段时长结束的时候发送源Observable发送的最新数据，其它的都会被过滤掉。也就是说在sample指定的这段时间内无论源Observable发送了多少数据，只有最后一个数据才能通过sample的过滤。
sample是取规定时间段内的最后一个数据，而throttleFirst操作符取的则是规定时间段内的第一个数据，其它的会被过滤掉。在Android开发中，经常会有过滤掉过多的点击事件的需求，使用throttleFirst就可以很容易地将过多的点击事件过滤掉。
private void sample_throttleFirst(){
Observable.interval(0,200, TimeUnit.MILLISECONDS)
.sample(1000, TimeUnit.MILLISECONDS)
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
System.out.println("sample: "+ aLong);
}
});
Observable.interval(0,200, TimeUnit.MILLISECONDS)
.throttleFirst(1000, TimeUnit.MILLISECONDS)
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
System.out.println("throttleFirst: "+ aLong);
}
});
}
结果：
throttleFirst: 0
是的， 第一个程序的结果没有输出，interval究竟是怎么用的呢，好像这本书的用法不对，也有可能是因为更新了库，这个后面再研究吧。
网友评论