看完不懂Rxjava我跪搓板（2）
2018-04-26 本文已影响173人
付凯强
0. 为了月薪1.8万
奋斗方向和目标.png
1. 序言
上一篇博客讲述内容如下：
- 函数响应式编程
- Rxjava概述
- Rxjava的优点
- Rxjava的组成
- Rxjava的原理
- Rxjava与观察者模式
- Rxjava的创建操作符
- Rxjava的变换操作符
如需了解，请点击链接进行跳转：https://www.jianshu.com/p/57b5c6567791
2. 过滤操作符
过滤操作符，顾名思义，就是对数据进行过滤，过滤掉我们不想要的，得到我们想要的。
- filter：对Observable产生的结果进行自定义规则地过滤，满足条件的才提交给订阅者。
Observable.just(1, 2, 3, 4).filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer > 2;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 21:32:40.319 11704-11704/? D/MainActivity: FILTER:3
04-26 21:32:40.319 11704-11704/? D/MainActivity: FILTER:4
- elementAt：用来返回指定位置的数据
Observable.just(1, 2, 3, 4).elementAt(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
Observable.just(1, 2, 3, 4).elementAt(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
- distinct：去重，只允许没有发射过的数据项通过
Observable.just(1, 2, 2, 3, 4).distinct().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 09:38:46.882 1786-1786/? D/MainActivity: FILTER:1
04-26 09:38:46.882 1786-1786/? D/MainActivity: FILTER:2
04-26 09:38:46.882 1786-1786/? D/MainActivity: FILTER:3
04-26 09:38:46.882 1786-1786/? D/MainActivity: FILTER:4
- distinctUntilChanged：去重，只过滤连续重复的数据
Observable.just(1, 2, 3,2, 3, 4).distinctUntilChanged().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 09:41:14.589 1964-1964/? D/MainActivity: FILTER:1
04-26 09:41:14.589 1964-1964/? D/MainActivity: FILTER:2
04-26 09:41:14.589 1964-1964/? D/MainActivity: FILTER:3
04-26 09:41:14.589 1964-1964/? D/MainActivity: FILTER:2
04-26 09:41:14.589 1964-1964/? D/MainActivity: FILTER:3
04-26 09:41:14.589 1964-1964/? D/MainActivity: FILTER:4
- skip：将Observalb发射的数据过滤掉前n项
Observable.just(1, 2, 3,2, 3, 4).skip(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 09:51:22.934 2073-2073/com.example.multidownload D/MainActivity: FILTER:3
04-26 09:51:22.935 2073-2073/com.example.multidownload D/MainActivity: FILTER:2
04-26 09:51:22.935 2073-2073/com.example.multidownload D/MainActivity: FILTER:3
04-26 09:51:22.935 2073-2073/com.example.multidownload D/MainActivity: FILTER:4
- take：只取前n项
Observable.just(1, 2, 3,2, 3, 4).take(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 09:52:55.417 2157-2157/? D/MainActivity: FILTER:1
04-26 09:52:55.418 2157-2157/? D/MainActivity: FILTER:2
- skipLast：过滤掉后n项
Observable.just(1, 2, 3,2, 3, 4).skipLast(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 09:54:46.296 2242-2242/? D/MainActivity: FILTER:1
04-26 09:54:46.296 2242-2242/? D/MainActivity: FILTER:2
04-26 09:54:46.296 2242-2242/? D/MainActivity: FILTER:3
04-26 09:54:46.296 2242-2242/? D/MainActivity: FILTER:2
- takeLast：只拿后n项
Observable.just(1, 2, 3,2, 3, 4).takeLast(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "FILTER:" + integer);
}
});
04-26 09:54:46.296 2242-2242/? D/MainActivity: FILTER:3
04-26 09:54:46.296 2242-2242/? D/MainActivity: FILTER:4
- ignoreElements：忽略源Observable发射的数据，只把Observable的onCompleted和onError事件通知给订阅者,也就是说只关心发射成功或失败，不关心发射内容
Observable.just(1, 2, 3, 2, 3, 4).ignoreElements().subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext");
}
});
04-26 10:09:33.294 2362-2362/com.example.multidownload D/MainActivity: onCompleted
- throttleFirst：定期发射这个时间里源Observable发射的第一个数据：
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for (int i = 0; i < 10; i++) {
subscriber.onNext(i);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
subscriber.onCompleted();
}
}).throttleFirst(200, TimeUnit.MILLISECONDS)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "INTEGER:" + integer);
}
});
04-26 11:02:57.708 2784-2784/? D/MainActivity: INTEGER:0
04-26 11:02:57.930 2784-2784/? D/MainActivity: INTEGER:2
04-26 11:02:58.135 2784-2784/? D/MainActivity: INTEGER:4
04-26 11:02:58.336 2784-2784/? D/MainActivity: INTEGER:6
04-26 11:02:58.543 2784-2784/? D/MainActivity: INTEGER:8
- throttleWithTimeOut：时间间隔限定：如果在这个时间段的数据就会被过滤掉。
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for (int i = 0; i < 10; i++) {
subscriber.onNext(i);
int sleep = 100;
if (i % 3 == 0) {
sleep = 300;
}
try {
Thread.sleep(sleep);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
subscriber.onCompleted();
}
}).throttleWithTimeout(200, TimeUnit.MILLISECONDS)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "INTEGER:" + integer);
}
});
04-26 11:19:35.525 3408-3422/com.example.multidownload D/MainActivity: INTEGER:0
04-26 11:19:36.030 3408-3422/com.example.multidownload D/MainActivity: INTEGER:3
04-26 11:19:36.541 3408-3422/com.example.multidownload D/MainActivity: INTEGER:6
04-26 11:19:37.037 3408-3422/com.example.multidownload D/MainActivity: INTEGER:9
3. 后续
如果大家喜欢这篇文章，欢迎点赞；如果想看更多前端移动端后端Java或Python方面的技术，欢迎关注！
