1、前言
-
RxJava凭借着其层次清晰、逻辑简洁，深受开发者的喜爱； - 那么要学习
RxJava，就必须掌握它的各类操作符； - 本篇文章对常用的转换操作符作一个归纳，以此来扩展对
RxJava的使用；
2、map()
2.1、作用
把要发送的事件类型转换成其他类型
2.2、具体使用
Observable.just(1,2,3).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return new StringBuffer().append("测试").append(integer).toString();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "onSubscribe");
}
@Override
public void onNext(String s) {
Log.i("测试", "onNext->" + s);
}
@Override
public void onError(Throwable e) {
Log.i("测试", "onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "onComplete");
}
});
2.3、打印结果
07-18 15:45:09.277 11483-11483/com.qinkl I/测试: onSubscribe
onNext->测试1
07-18 15:45:09.278 11483-11483/com.qinkl I/测试: onNext->测试2
onNext->测试3
onComplete
3、flatMap()
3.1、作用
flatMap和map的区别在于，flatMap返回的是Observable，并且对数据进行批量的处理；个人认为，好像嵌套多层的for循环;
3.2、具体使用
ArrayList<String> chinese = new ArrayList<String>();
chinese.add("小学语文");
chinese.add("初中语文");
chinese.add("大学语文");
ArrayList<String> math = new ArrayList<String>();
math.add("小学数学");
math.add("初中数学");
math.add("大学数学");
ArrayList<Course> course = new ArrayList<Course>();
course.add(new Course("语文",chinese));
course.add(new Course("数学",math));
List<Student> students = new ArrayList<>();
students.add(new Student("张三",course));
students.add(new Student("李四",course));
Observable.fromIterable(students)
.flatMap(new Function<Student, ObservableSource<Course>>() {
@Override
public ObservableSource<Course> apply(Student student) throws Exception {
return Observable.fromIterable(student.getCourseList());
}
})
.flatMap(new Function<Course, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Course course) throws Exception {
return Observable.fromIterable(course.getActionList());
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "onSubscribe");
}
@Override
public void onNext(String s) {
Log.i("测试", "onNext->" + s);
}
@Override
public void onError(Throwable e) {
Log.i("测试", "onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "onComplete");
}
});
}
3.3、打印结果
07-18 16:12:56.266 12805-12805/com.qinkl I/测试: onSubscribe
07-18 16:12:56.267 12805-12805/com.qinkl I/测试: onNext->小学语文
onNext->初中语文
onNext->大学语文
onNext->小学数学
onNext->初中数学
onNext->大学数学
onNext->小学语文
07-18 16:12:56.268 12805-12805/com.qinkl I/测试: onNext->初中语文
onNext->大学语文
onNext->小学数学
onNext->初中数学
onNext->大学数学
onComplete
4、concatMap()
和flatMap()差不多，区别在于flatMap()是无序的，concatMap()是有序的；
5、buffer()
5.1、作用
缓存一定的事件统一发送，直到事件发送完；
5.2、具体使用
Observable.just(1,2,3,4,5,6)
.buffer(3,1)
.subscribe(new Observer<List<Integer>>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "onSubscribe");
}
@Override
public void onNext(List<Integer> s) {
Log.i("测试", "onNext->缓存大小:" + s.size());
for (int j = 0; j < s.size(); j++) {
Log.i("测试", "事件:" + s.get(j));
}
}
@Override
public void onError(Throwable e) {
Log.i("测试", "onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "onComplete");
}
});
}
buffer()有count和skip两个参数，count表示缓存区大小，skip表示下一次发送事件要跳过的事件数量；比如第一次缓存3个：1、2、3；假如skip为1，那么下一次发送缓冲事件为跳过1事件，也就是2事件，从2开始发送；
5.3、打印结果
07-18 16:47:08.405 15292-15292/com.qinkl I/测试: onSubscribe
onNext->缓存大小:3
07-18 16:47:08.406 15292-15292/com.qinkl I/测试: 事件:1
事件:2
事件:3
onNext->缓存大小:3
事件:2
事件:3
事件:4
onNext->缓存大小:3
07-18 16:47:08.407 15292-15292/com.qinkl I/测试: 事件:3
事件:4
事件:5
onNext->缓存大小:3
事件:4
事件:5
事件:6
onNext->缓存大小:2
事件:5
事件:6
onNext->缓存大小:1
事件:6
07-18 16:47:08.408 15292-15292/com.qinkl I/测试: onComplete
6、groupBy()
6.1、作用
对事件进行指定条件的分组，按分组发送事件，有点抽象，具体看例子和打印结果；
6.2、具体使用
Observable.just(3,6,9,4,7,8,10)
.groupBy(new Function<Integer, Integer>() {
@Override
public Integer apply(Integer integer) throws Exception {
return integer % 3;
}
})
.subscribe(new Observer<GroupedObservable<Integer,Integer>>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "onSubscribe");
}
@Override
public void onNext(final GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) {
Log.i("测试", "onNext");
integerIntegerGroupedObservable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "integerIntegerGroupedObservable-onSubscribe");
}
@Override
public void onNext(Integer integer) {
Log.i("测试", "integerIntegerGroupedObservable-onNext:value:"+integer+",groupName:"+integerIntegerGroupedObservable.getKey());
}
@Override
public void onError(Throwable e) {
Log.i("测试", "integerIntegerGroupedObservable-onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "integerIntegerGroupedObservable-onComplete");
}
});
}
@Override
public void onError(Throwable e) {
Log.i("测试", "onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "onComplete");
}
});
}
6.3、打印结果
07-18 17:16:58.748 17290-17290/com.qinkl I/测试: onSubscribe
07-18 17:16:58.750 17290-17290/com.qinkl I/测试: onNext
integerIntegerGroupedObservable-onSubscribe
07-18 17:16:58.751 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onNext:value:3,groupName:0
07-18 17:16:58.752 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onNext:value:6,groupName:0
integerIntegerGroupedObservable-onNext:value:9,groupName:0
onNext
07-18 17:16:58.753 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onSubscribe
integerIntegerGroupedObservable-onNext:value:4,groupName:1
07-18 17:16:58.754 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onNext:value:7,groupName:1
onNext
integerIntegerGroupedObservable-onSubscribe
07-18 17:16:58.754 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onNext:value:8,groupName:2
07-18 17:16:58.755 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onNext:value:10,groupName:1
07-18 17:16:58.756 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onComplete
07-18 17:16:58.756 17290-17290/com.qinkl I/测试: integerIntegerGroupedObservable-onComplete
onComplete
7、scan()
7.1、作用
按特定逻辑对事件进行聚合处理
7.2、具体使用
Observable.just(1,2,3,4,5)
.scan(new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(Integer integer, Integer integer2) throws Exception {
Log.i("测试", "apply:"+integer+","+integer2);
return integer+integer2;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "onSubscribe");
}
@Override
public void onNext(Integer s) {
Log.i("测试", "onNext:" + s);
}
@Override
public void onError(Throwable e) {
Log.i("测试", "onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "onComplete");
}
});
}
7.3、打印结果
07-18 17:40:20.512 18130-18130/com.qinkl I/测试: onSubscribe
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: onNext:1
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: apply:1,2
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: onNext:3
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: apply:3,3
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: onNext:6
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: apply:6,4
onNext:10
apply:10,5
onNext:15
onComplete
8、window()
8.1、作用
按照数量对事件进行分组
8.2、具体使用
Observable.just(1,2,3,4,5,6,7)
.window(3)
.subscribe(new Observer<Observable<Integer>>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("测试", "onSubscribe");
}
@Override
public void onNext(Observable<Integer> integerObservable) {
Log.i("测试", "onNext:");
integerObservable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.i("测试", "integerObservable-onNext:"+integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
@Override
public void onError(Throwable e) {
Log.i("测试", "onError" + e);
}
@Override
public void onComplete() {
Log.i("测试", "onComplete");
}
});
}
8.3、打印结果
07-18 17:40:20.512 18130-18130/com.qinkl I/测试: onSubscribe
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: onNext:1
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: apply:1,2
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: onNext:3
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: apply:3,3
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: onNext:6
07-18 17:40:20.513 18130-18130/com.qinkl I/测试: apply:6,4
onNext:10
apply:10,5
onNext:15
onComplete
9、总结
map()：把要发送的事件类型转换成其他类型
flatMap()：flatMap和map的区别在于，flatMap返回的是Observable，并且对数据进行批量的处理；
concatMap()：和flatMap()差不多，区别在于flatMap()是无序的，concatMap()是有序的；
buffer()：缓存一定的事件统一发送，直到事件发送完；
groupBy()：对事件进行指定条件的分组，按分组发送事件，有点抽象，具体看例子和打印结果；
scan()：按特定逻辑对事件进行聚合处理
window()：按照数量对事件进行分组
