1.filter
根据Predicate类中,给定的条件,对上游集合中的数据进行判断。如果给定条件返回true则继续向下传递,否则过滤掉。
-
- Observable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9).filter(new Predicate<Integer>() {
- @Override
- public boolean test(@NonNull Integer integer) throws Exception {
- return integer % 2 == 0;
- }
- }).subscribe(new Observer<Integer>() {
- @Override
- public void onSubscribe(@NonNull Disposable d) {
-
- }
-
- @Override
- public void onNext(@NonNull Integer integer) {
- Log.e("rxjava","onNext "+integer);
- }
-
- @Override
- public void onError(@NonNull Throwable e) {
-
- }
-
- @Override
- public void onComplete() {
-
- }
- });
- 执行结果:
- rxjava: onNext 2
- rxjava: onNext 4
- rxjava: onNext 6
- rxjava: onNext 8
上面的列子是一个数组,给定条件是integer % 2 == 0,如果是偶数则继续传递,否则过滤掉。
filter返回的是ObservableFilter,ObservableFilter(ObservableSource
接收两个参数,source就是传进来的被观察者,predicate就是预置条件,如果为true,则调用下游的onNext方法进行传递。
2.distinct:不同的
去重,去掉被观察者中重复的元素
- Observable.fromArray(1,2,3,4,5,4,3,2,1).distinct().subscribe(new Observer
() { - @Override
- public void onSubscribe(@NonNull Disposable d) {
-
- }
-
- @Override
- public void onNext(@NonNull Integer integer) {
- Log.e("rxjava","onNext "+integer);
- }
-
- @Override
- public void onError(@NonNull Throwable e) {
-
- }
-
- @Override
- public void onComplete() {
-
- }
- });
- 执行结果:
- rxjava: onNext 1
- rxjava: onNext 2
- rxjava: onNext 3
- rxjava: onNext 4
- rxjava: onNext 5
distinct去掉集合中重复的元素。通过源码分析可知,去重原理是通过创建的HashSet去重的。
- public final Observable
distinct() { - return distinct(Functions.identity(), Functions.createHashSet());
- }
3.skip:过滤掉前面的n项数据
skipLast:过滤掉后面的n项数据
- Observable.fromArray(1,2,3,4,5,6,7,8,9).skip(3).skipLast(3).subscribe(new Observer
() { - @Override
- public void onSubscribe(@NonNull Disposable d) {
-
- }
-
- @Override
- public void onNext(@NonNull Integer integer) {
- Log.e("rxjava","onNext "+integer);
- }
-
- @Override
- public void onError(@NonNull Throwable e) {
-
- }
-
- @Override
- public void onComplete() {
-
- }
- });
- 执行结果:
- rxjava: onNext 4
- rxjava: onNext 5
- rxjava: onNext 6
4.take 取前n项数据
take 取后n项数据
- Observable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9).take(3).subscribe(new Observer
() { - @Override
- public void onSubscribe(@NonNull Disposable d) {
-
- }
-
- @Override
- public void onNext(@NonNull Integer integer) {
- Log.e("rxjava", "onNext " + integer);
- }
-
- @Override
- public void onError(@NonNull Throwable e) {
-
- }
-
- @Override
- public void onComplete() {
-
- }
- });
- 执行结果:
- rxjava: onNext 1
- rxjava: onNext 2
- rxjava: onNext 3