RxJava过滤操作符实例

过滤操作符

[b]filter[/b]

符合某种规则的Observable才会向下传递，例子

Observable.range(100,10).filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer>105;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {

}
});

上面的Subscriber只会得到大于105的数值

[b]first[/b]

只发射第一项，或者是满足条件的第一项，两个例子

Observable.range(100,5).first();// 发射100
Observable.range(100,5).first(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer>102;//发射103
}
});

[b]Debounce[/b]

当操作太频繁的时候，在第N个节点发射之前会忽略钱N-1个节点，比如我们的Edittext去seach的时候，可能需要过滤前多少秒的操作只是需要后续的操作。简单点来说，就是，他需要的是在某个时间点之后的结果，在这个时间点之前的结果他会忽略不再发送给下一级。例子：

Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
try {
for (int i = 1; i <= 10; i++) {
subscriber.onNext(i);
Thread.sleep(100*i);
}
subscriber.onCompleted();
} catch (InterruptedException e) {
e.printStackTrace();
subscriber.onError(e);
}
}
}).subscribeOn(Schedulers.newThread())
.debounce(500, TimeUnit.MILLISECONDS).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.i(TAG, "OnComplete");
}

@Override
public void onError(Throwable e) {
Log.i(TAG, "Throwable");
}

@Override
public void onNext(Integer integer) {
Log.i(TAG, "integer == "+integer.toString());
}
});

输出就是6.7.8.9.10

总结就是debounce操作符就是类似一个控制入口，他是有一个门卡的，门卡一般是时间，在这之前的忽略，之后的可以进入

例子是来自 Android RxJava使用介绍（三） RxJava的操作符

其中，还有一个throttleWithTimeout操作符，他的底层也是通过调用debounce操作符完成的，作用类似。但是debounce比她在、多了一个功能就是可以通过函数来进行限流操作。例子如下：

Observable.just(1, 2, 3, 4, 5).debounce(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
return Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
if (integer % 2 == 0 && !subscriber.isUnsubscribed()) {
subscriber.onNext(integer);
subscriber.onCompleted();
}
}
});
}
}).subscribeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==integer");
}
});

结果输出是2.4.5。5为什么在其中？因为，在debounce操作符中，最后一个Observable，在被调用onComplete之前的时间间隔内被调用，那么他也会被传递到下一级。







[b]Distinct[/b]

只允许没有发射过的数据发射，可以定义过滤规则，例子

Observable.range(1,5).distinct(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
if(integer<=2) return "1";// 定义过滤规则
return "2";
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString());
}
});
Observable.just(1,2,3,1,2).distinct().subscribe(new Action1<Integer>() { // 使用默认规则
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString());
}
});

输出

09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 1==自定义
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 3==自定义
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 1===系统
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 2===系统
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 3===系统

[b]elementAt[/b]

某一位置的元素可以发射，需要需要越界的问题，他是从0开始计算的例子

Observable.range(10,10).elementAt(9).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {

}
});

[b]last[/b]

最后一个或者满足条件的最后一个，跟first相反，例子

Observable.range(10,5).last();
Observable.range(10,5).last(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer>13;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString()+"===last");
}
});

[b]skip[/b]

跳过前N项，从第N+1开始发射，例子

Observable.range(10,20).skip(10);

这个例子是丢弃了10~29中的10~19，保留20~29，还有两个重载的方法

skip(long,TimeUnit));// 跳过前long时间的Observable，发射之后的
skip(long,TimeUnit,Scheduler));//可以定义调度器

结果是14，两个都是





[b]skipLast[/b]

不发射后n项数据，跟skip相反，例如

Observable.range(1,10).skipLast(3).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==skip");
}
});

输出：1,2,3,4,5,6,7



[b]take[/b]

只取前n项数据，例如有10项数据，那么take(4)，就是只是圈前4项数据，例如

Observable.range(10,10).take(5).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==take");
}
});

结果是：10,11，12,13,14



[b]taskLast[/b]

只发射最后n项数据，比如有10项数据，taskLast(3)，则是只发射最后3项数据，例子：

Observable.range(9,5).takeLast(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==integer");
}
});

输出：12和13

需要注意的是他是一个线程同步的，他可能造成Observable发射延迟，因为他只有过滤掉前面的数据才会轮到后面，然后后面的数据才会发射。

重构的方法有：

takeLast(final int count)  //发射后面n个数据，n大于数据全部的时候则发射全部n小于0报异常
takeLast(int count, long time, TimeUnit unit)// 发射最后给定时间的n个数据
takeLast(int count, long time, TimeUnit unit, Scheduler scheduler)
takeLast(long time, TimeUnit unit, Scheduler scheduler)//发射最后给定时间的全部数据，可以指定调度器
takeLast(long time, TimeUnit unit)

同时，他还有一个变形，takeLastBuffer*，takeLast是一个一个数据源返回，但是takeLastBuffer是以List的形式返回，例如：

Observable.range(20,10).takeLastBuffer(5).subscribe(new Action1<List<Integer>>() {
@Override
public void call(List<Integer> integers) {
Log.i(TAG, integers.toString() + "==integer");
}
});

[b]ignoreElements[/b]

忽略所有的元素，只是发射结束事件或者是错误时间，例如

Observable.range(10,10).ignoreElements().subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError");
}
@Override
public void onNext(Integer integer) {
Log.i(TAG, integer.toString() + "==ignoreElements");
}
});

[b]last[/b]

只发射最后的一项数据，忽略前面n项数据，例如

Observable.range(10,10).last().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==last");
}
});

结果是：19



还有一个last(Func1)的重载函数，他可以规定那个才是last，比如

Observable.range(10,10).last(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
Log.i(TAG, integer.toString() + "==last");
return integer == 14;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==last func");
}
});

在subscribe里面输出的就是14而不是19，还有一个相似函数是:lastOrDefault(T)，他会在Observable没有发射任何数据的时候发射默认值T。



[b]sample操作符，[/b]

他是在某一段时间内采集过去发射的旧数据再次发射，比如：

Observable.interval(1,TimeUnit.SECONDS).sample(2,TimeUnit.SECONDS).subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
Log.i(TAG, aLong.toString() + "==sample ");
}
});

interval创造一个Observable之后每个1s发射一个新数据，然后sample会在2s重复的去采集已经发射的旧数据再次发射，输出的结果是:0,2,4,6…

它默认开启新线程，我们也可指线程给sample运行

sample例子来自RxJava 过滤操作符 sample