Android ：RxJava学习笔记之过滤操作符
Posted JMW1407
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Android ：RxJava学习笔记之过滤操作符相关的知识，希望对你有一定的参考价值。
RxJava ——过滤操作符
过滤操作符
顾名思义，这类operators主要用于对事件数据的筛选过滤，只返回满足我们条件的数据。
分类；
- 1、根据 指定条件 过滤事件
- 2、根据 指定事件数量 过滤事件
- 3、根据 指定时间 过滤事件
- 4、根据 指定事件位置 过滤事件
1、根据 指定条件 过滤事件
1.1、Filter（）
filter(Func1)用来过滤观测序列中我们不想要的值，只返回满足条件的值，我们看下原理图：
Observable
.create((ObservableOnSubscribe<Integer>) emitter -> {
// 1. 发送5个事件
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onNext(5);
}
)
.filter(integer -> {
// 根据test()的返回值 对被观察者发送的事件进行过滤 & 筛选
// a. 返回true，则继续发送
// b. 返回false，则不发送（即过滤）
return integer > 3;
// 本例子 = 过滤了整数≤3的事件
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable disposable) {
System.out.println("对onSubscribe事件作出响应");
}
@Override
public void onNext(Integer value) {
System.out.println("接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
System.out.println("对Error事件作出响应" );
}
@Override
public void onComplete() {
System.out.println("对Complete事件作出响应");
}
});
输出
对onSubscribe事件作出响应
接收到了事件4
接收到了事件5
1.2、ofType（）
过滤 特定数据类型的数据
Observable.just(1, "Carson", 3, "Ho", 5)
.ofType(Integer.class) // 筛选出 整型数据
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"获取到的整型事件元素是： "+ integer);
}
});
1.3、skip（） / skipLast（）
skip(int)让我们可以忽略Observable发射的前n项数据。
SkipLast：skipLast(int)忽略Observable发射的后n项数据。
// 使用1：根据顺序跳过数据项
Observable.just(1, 2, 3, 4, 5)
.skip(1) // 跳过正序的前1项
.skipLast(2) // 跳过正序的后2项
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
System.out.println("Seq 获取到的整型事件元素是： "+ integer);
}
});
// 使用2：根据时间跳过数据项
// 发送事件特点：发送数据0-5，每隔1s发送一次，每次递增1；第1次发送延迟0s
Observable.intervalRange(0, 5, 0, 1, TimeUnit.SECONDS)
.skip(1, TimeUnit.SECONDS) // 跳过第1s发送的数据
.skipLast(1, TimeUnit.SECONDS) // 跳过最后1s发送的数据
.subscribe(new Consumer<Long>() {
@Override
public void accept( Long along ) throws Exception {
System.out.println("Time 获取到的整型事件元素是： "+ along);
}
});
try {
Thread.sleep(8000);
} catch (InterruptedException e) {
e.printStackTrace();
}
1.4、distinct（） / distinctUntilChanged（）
过滤事件序列中重复的事件 / 连续重复的事件
distinct()的过滤规则是只允许还没有发射过的数据通过，所有重复的数据项都只会发射一次。
distinct(Func1)参数中的Func1中的call方法会根据Observable发射的值生成一个Key，然后比较这个key来判断两个数据是不是相同；如果判定为重复则会和distinct()一样过滤掉重复的数据项。
distinctUntilChanged()和distinct()类似，只不过它判定的是Observable发射的当前数据项和前一个数据项是否相同。
// 使用1：过滤事件序列中重复的事件
Observable.just(1, 2, 3, 1 , 2 )
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"不重复的整型事件元素是： "+ integer);
}
});
// 使用2：过滤事件序列中 连续重复的事件
// 下面序列中，连续重复的事件 = 3、4
Observable.just(1,2,3,1,2,3,3,4,4 )
.distinctUntilChanged()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"不连续重复的整型事件元素是： "+ integer);
}
});
2、根据 指定事件数量 过滤事件
通过设置指定的事件数量，仅发送特定数量的事件
2.1、take（）
take(int)用一个整数n作为一个参数，从原始的序列中发射前n个元素.
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 1. 发送5个事件
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onNext(5);
}
// 采用take（）变换操作符
// 指定了观察者只能接收2个事件
}).take(2)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "过滤后得到的事件是："+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
// 实际上，可理解为：被观察者还是发送了5个事件，只是因为操作符的存在拦截了3个事件，最终观察者接收到的是2个事件
2.2、TakeLast
takeLast(int)同样用一个整数n作为参数，只不过它发射的是观测序列中后n个元素。
Observable.just(1, 2, 3, 4, 5)
.takeLast(3) //指定观察者只能接受被观察者发送的3个事件
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "过滤后得到的事件是："+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
3、根据 指定时间 过滤事件
通过设置指定的时间，仅发送在该时间内的事件
3.1、throttleFirst（）/ throttleLast（）
在某段时间内，只发送该段时间内第1次事件 / 最后1次事件
- 如，1段时间内连续点击按钮，但只执行第1次的点击操作
<<- 在某段时间内，只发送该段时间内第1次事件 ->>
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 隔段事件发送时间
e.onNext(1);
Thread.sleep(500);
e.onNext(2);
Thread.sleep(400);
e.onNext(3);
Thread.sleep(300);
e.onNext(4);
Thread.sleep(300);
e.onNext(5);
Thread.sleep(300);
e.onNext(6);
Thread.sleep(400);
e.onNext(7);
Thread.sleep(300);
e.onNext(8);
Thread.sleep(300);
e.onNext(9);
Thread.sleep(300);
e.onComplete();
}
}).throttleFirst(1, TimeUnit.SECONDS)//每1秒中采用数据
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
<<- 在某段时间内，只发送该段时间内最后1次事件 ->>
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 隔段事件发送时间
e.onNext(1);
Thread.sleep(500);
e.onNext(2);
Thread.sleep(400);
e.onNext(3);
Thread.sleep(300);
e.onNext(4);
Thread.sleep(300);
e.onNext(5);
Thread.sleep(300);
e.onNext(6);
Thread.sleep(400);
e.onNext(7);
Thread.sleep(300);
e.onNext(8);
Thread.sleep(300);
e.onNext(9);
Thread.sleep(300);
e.onComplete();
}
}).throttleLast(1, TimeUnit.SECONDS)//每1秒中采用数据
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
3.2、Sample（）
在某段时间内，只发送该段时间内最新（最后）1次事件,与 throttleLast（） 操作符类似,仅需要把上文的 throttleLast（） 改成Sample（）操作符即可
3.3、throttleWithTimeout （） / debounce（）
发送数据事件时，若2次发送事件的间隔＜指定时间，就会丢弃前一次的数据，直到指定时间内都没有新数据发射时才会发送后一次的数据
debounce(long, TimeUnit)过滤掉了由Observable发射的速率过快的数据；如果在一个指定的时间间隔过去了仍旧没有发射一个，那么它将发射最后的那个。通常我们用来结合RxBing(Jake Wharton大神使用RxJava封装的android UI组件)使用，防止button重复点击
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 隔段事件发送时间
e.onNext(1);
Thread.sleep(500);
e.onNext(2); // 1和2之间的间隔小于指定时间1s，所以前1次数据（1）会被抛弃，2会被保留
Thread.sleep(1500); // 因为2和3之间的间隔大于指定时间1s，所以之前被保留的2事件将发出
e.onNext(3);
Thread.sleep(1500); // 因为3和4之间的间隔大于指定时间1s，所以3事件将发出
e.onNext(4);
Thread.sleep(500); // 因为4和5之间的间隔小于指定时间1s，所以前1次数据（4）会被抛弃，5会被保留
e.onNext(5);
Thread.sleep(500); // 因为5和6之间的间隔小于指定时间1s，所以前1次数据（5）会被抛弃，6会被保留
e.onNext(6);
Thread.sleep(1500); // 因为6和Complete实践之间的间隔大于指定时间1s，所以之前被保留的6事件将发出
e.onComplete();
}
}).throttleWithTimeout(1, TimeUnit.SECONDS)//每1秒中采用数据
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
4、根据 指定事件位置 过滤事件
通过设置指定的位置，过滤在该位置的事件
4.1、firstElement（） / lastElement（）
仅选取第1个元素 / 最后一个元素
// 获取第1个元素
Observable.just(1, 2, 3, 4, 5)
.firstElement()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"获取到的第一个事件是： "+ integer);
}
});
// 获取最后1个元素
Observable.just(1, 2, 3, 4, 5)
.lastElement()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"获取到的最后1个事件是： "+ integer);
}
});
4.2、elementAt（）
elementAt(int)用来获取元素Observable发射的事件序列中的第n项数据，并当做唯一的数据发射出去。
// 使用1：获取位置索引 = 2的 元素
// 位置索引从0开始
Observable.just(1, 2, 3, 4, 5)
.elementAt(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"获取到的事件元素是： "+ integer);
}
});
// 使用2：获取的位置索引 ＞ 发送事件序列长度时，设置默认参数
Observable.just(1, 2, 3, 4, 5)
.elementAt(6,10)
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"获取到的事件元素是： "+ integer);
}
});
4.3、elementAtOrError（）
在elementAt（）的基础上，当出现越界情况（即获取的位置索引 ＞ 发送事件序列长度）时，即抛出异常
Observable.just(1, 2, 3, 4, 5)
.elementAtOrError(6)
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"获取到的事件元素是： "+ integer);
}
});
参考
以上是关于Android ：RxJava学习笔记之过滤操作符的主要内容，如果未能解决你的问题，请参考以下文章