过滤操作符
-
作用
过滤 / 筛选 被观察者（Observable）发送的事件 & 观察者 （Observer）接收的事件
-
类型
类型图.png
-
应用场景 & 对应操作符详解
1.Filter（）
作用
过滤 特定条件的事件。
实例
public void FilterOperators(View view){
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onNext(5);
}
}).filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer>3;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e("sss", "开始采用subscribe连接");
}
@Override
public void onNext(Integer integer) {
Log.e("sss", "过滤后得到的事件是："+ integer );
}
@Override
public void onError(Throwable e) {
Log.e("sss", "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.e("sss", "对Complete事件作出响应");
}
});
}
2.ofType（）
作用
过滤 特定数据类型的数据
事例
public void ofTypeOperators(View view){
Observable.just(1,"hh",6,"lll",8)
.ofType(Integer.class)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的整型事件元素是： "+ integer);
}
});
}
3.skip（） / skipLast（）
作用
跳过某个事件
事例
public void skipOperators(View view){
Observable.just(1,2,3,4,5)
.skip(1)
.skipLast(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的整型事件元素是： "+ integer);
}
});
}
4.distinct（） / distinctUntilChanged（）
作用
过滤事件序列中重复的事件 / 连续重复的事件
事例
public void distinctOperators(View view){
Observable.just(1,2,5,1,5,6,3)
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的不重复事件元素是： "+ integer);
}
});
}
5.take（）
作用
指定观察者最多能接收到的事件数量
事例
public void takeOperators(View view){
Observable.just(1,2,3,4,6)
.take(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的事件元素是： "+ integer);
}
});
}
6.takeLast（）
作用
指定观察者只能接收到被观察者发送的最后几个事件
事例
public void takeLastOperators(View view){
Observable.just(1,2,3,4,5)
.takeLast(3)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e("sss", "开始采用subscribe连接");
}
@Override
public void onNext(Integer integer) {
Log.e("sss", "接受后得到的事件是："+ integer );
}
@Override
public void onError(Throwable e) {
Log.e("sss", "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.e("sss", "对Complete事件作出响应");
}
});
}
7.throttleFirst（）/ throttleLast（）
作用
在某段时间内，只发送该段时间内第1次事件 / 最后1次事件
事例
public void throttleFirstOperators(View view){
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
Thread.sleep(500);
emitter.onNext(2);
Thread.sleep(400);
emitter.onNext(3);
Thread.sleep(300);
emitter.onNext(4);
Thread.sleep(300);
emitter.onNext(5);
Thread.sleep(300);
emitter.onNext(6);
Thread.sleep(400);
emitter.onNext(7);
Thread.sleep(300);
emitter.onNext(8);
Thread.sleep(300);
emitter.onNext(9);
Thread.sleep(300);
emitter.onComplete();
}
}).throttleFirst(1, TimeUnit.SECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e("sss", "开始采用subscribe连接");
}
@Override
public void onNext(Integer integer) {
Log.e("sss", "接收到了事件"+ integer );
}
@Override
public void onError(Throwable e) {
Log.e("sss", "对Error事件作出响应" );
}
@Override
public void onComplete() {
Log.e("sss","对Complete事件作出响应");
}
});
}
8.Sample（）
作用
在某段时间内，只发送该段时间内最新（最后）1次事件与 throttleLast（） 操作符类似。
9.throttleWithTimeout （） / debounce（）
作用
发送数据事件时，若2次发送事件的间隔＜指定时间，就会丢弃前一次的数据，直到指定时间内都没有新数据发射时才会发送后一次的数据。
事例
public void throttleWithTimeoutOperators(View view){
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
Thread.sleep(500);
emitter.onNext(2);
Thread.sleep(1500);
emitter.onNext(3);
Thread.sleep(1500);
emitter.onNext(4);
Thread.sleep(500);
emitter.onNext(5);
Thread.sleep(1500);
emitter.onNext(6);
Thread.sleep(500);
emitter.onComplete();
}
}).throttleWithTimeout(1,TimeUnit.MICROSECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.e("sss", "接收到了事件"+ integer );
}
@Override
public void onError(Throwable e) {
Log.e("sss", "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.e("sss", "对Complete事件作出响应");
}
});
}
- firstElement（） / lastElement（）
作用
仅选取第1个元素 / 最后一个元素
事例
public void firstElementOperators(View view){
Observable.just(1,2,3,4,5)
.firstElement()
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的第一个事件是： "+ integer);
}
});
}
11.elementAt（）
作用
指定接收某个元素（通过 索引值 确定）。
事例
public void elementAtOperators(View view){
Observable.just(1,2,3,4,5)
.elementAt(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的事件元素是： "+ integer);
}
});
}
12.elementAtOrError（）
作用
在elementAt（）的基础上，当出现越界情况（即获取的位置索引 ＞ 发送事件序列长度）时，即抛出异常。
事例
public void elementAtOrErrorOperators(View view){
Observable.just(1,2,3,4,5)
.elementAtOrError(6)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("sss","获取到的事件元素是： "+ integer);
}
});
}
-
实际开发中的应用
1.功能防抖
throttleFirst的使用
Button bt=findViewById(R.id.bt);
RxView.clicks(bt)
.throttleFirst(4, TimeUnit.SECONDS)
.subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object o) {
Log.e("sss", "发送了网络请求" );
}
@Override
public void onError(Throwable e) {
Log.e("sss", "对Error事件作出响应" + e.toString());
}
@Override
public void onComplete() {
Log.e("sss", "对Complete事件作出响应");
}
});
}
