当你有一个需要订阅的Observable,并且希望转换结果的时候(切记,响应式编程中一切皆流)。即将涉及到observable转换的时候,从队列中取出将要消费的事件,不可能一直是我们需要的格式或者形状,可能每个值都需要扩展成更丰富的对象或者化作更多的值。为了达到目的,我们可以为每一个observable的返回值使用一个这样的方法函数,使用它可以将所有已发送的事件转换成各种Observable,并最终合并结果。
- map
map操作符的作用就是将一个事件转换成另一个事件,通过使用map中的方法对Observable中发射出来的所有数据进行变换,map可以进行多次变化,例如.map() .map()........
static List<Student> studentList = new ArrayList<Student>() {
{
add(new Student("科比", 24));
add(new Student("姚明", 11));
}
};
static Map<String, Course> courseMap = new HashMap<>();
static {
courseMap.put("科比", new Course("湖人", 1996));
courseMap.put("姚明", new Course("火箭", 1996));
}
//map
private static void testMap() {
final List<String> nameList = new ArrayList<>();
Observable.from(studentList).map(new Func1<Student, String>() {
@Override
public String call(Student student) {
return student.name;
}
}).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted nameList.size() = " + nameList.size());
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String value) {
System.out.println("onSuccess value = " + value);
nameList.add(value);
}
});
}
运行结果: 得到多个Student对象中的name,保存到nameList中
2.Flatmap
flatMap()究竟是用来干嘛的?简单的说,flatmap把一个Observable变成多个Observable,然后把得到的多个Obervable的元素一个个的发射出去。
在这里引用大神的例子进行理解:
首先假设这么一种需求:假设有一个数据结构『学生』,现在需要打印出一组学生的名字。实现方式很简单:
Student[] students = ...;
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onNext(String name) {
Log.d(tag, name);
}
...
};
Observable.from(students)
.map(new Func1<Student, String>() {
@Override
public String call(Student student) {
return student.getName();
}
})
.subscribe(subscriber);
那么再假设:如果要打印出每个学生所需要修的所有课程的名称呢?(需求的区别在于,每个学生只有一个名字,但却有多个课程。)首先可以这样实现:
Student[] students = ...;
Subscriber<Student> subscriber = new Subscriber<Student>() {
@Override
public void onNext(Student student) {
List<Course> courses = student.getCourses();
for (int i = 0; i < courses.size(); i++) {
Course course = courses.get(i);
Log.d(tag, course.getName());
}
}
...
};
Observable.from(students)
.subscribe(subscriber);
flatmap登场
Student[] students = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() {
@Override
public void onNext(Course course) {
Log.d(tag, course.getName());
}
...
};
Observable.from(students)
.flatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student student) {
return Observable.from(student.getCourses());
}
})
.subscribe(subscriber);
从上面的代码可以看出, flatMap() 和 map() 有一个相同点:它也是把传入的参数转化之后返回另一个对象。但需要注意,和 map() 不同的是, flatMap() 中返回的是个 Observable 对象,并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。 flatMap() 的原理是这样的:1. 使用传入的事件对象创建一个 Observable 对象;2. 并不发送这个 Observable, 而是将它激活,于是它开始发送事件;3. 每一个创建出来的 Observable 发送的事件,都被汇入同一个 Observable ,而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
3.ConcatMap操作符
类似于最简单版本的flatMap,但是它按次序连接而不是合并那些生成的Observables,然后产生自己的数据序列。
区别:无序:FlatMap对这些Observables发射的数据做的是合并(merge)操作,因此它们可能是交错的。
有序:ConcatMap不会让变换后的Observables发射的数据交错,它按照严格的顺序发射这些数据。
说明:在同步线程中,FlatMap和ConcactMap的执行结果是一样的(结果是有序的)
只有在异步线程中,FlatMap结果可能是无序的,而ConcactMap始终能保持有序的结果。
public static void test() {
List<Integer> integers = Arrays.asList(2, 3, 4, 5, 6, 7, 8, 9);
Observable.from(integers).flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
return Observable.just(integer).subscribeOn(Schedulers.from(Executors.newCachedThreadPool()));
}
}).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
System.out.println("flatMap onSuccess value = " + integer);
}
});
System.out.println("----------------------------");
Observable.from(integers)
.concatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
return Observable.just(integer).subscribeOn(Schedulers.from(Executors.newCachedThreadPool()));
}
})
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
System.out.println("concatMap onNext value = " + integer);
}
});
}
----------------------------
flatMap onSuccess value = 2
flatMap onSuccess value = 4
flatMap onSuccess value = 6
flatMap onSuccess value = 7
flatMap onSuccess value = 3
flatMap onSuccess value = 9
flatMap onSuccess value = 8
flatMap onSuccess value = 5
以上是乱序的
concatMap onNext value = 2
concatMap onNext value = 3
concatMap onNext value = 4
concatMap onNext value = 5
concatMap onNext value = 6
concatMap onNext value = 7
concatMap onNext value = 8
concatMap onNext value = 9
4.switchMap
解释:将Observable发射的数据集合变换为Observables集合,然后只发射这些Observables最近发射的数据
用法与FlatMap几乎一样,区别是SwitchMap操作符只会发射最近的Observables。当原Observable发射一个新的数据项时,如果旧数据项订阅还未完成,就取消旧订阅数据和停止监视那个数据项产生的Observable,开始监视新的数据项.
private static void test7() {
Observable.just("A", "B", "C", "D", "E")
.switchMap(new Func1<String, Observable<String>>() {
@Override
public Observable<String> call(String s) {
return Observable.just(s + "1").subscribeOn(Schedulers.newThread()); //并发
//return Observable.just(s+"1");
}
})
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
System.out.println("switchMap onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println("switchMap onError :" + e);
}
@Override
public void onNext(String s) {
System.out.println("switchMap Next :" + s);
}
});
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
逻辑推演:
* A --> 取消空的,没有可以取消的
* B--> A1被取消
* C--> B1被取消
* D--> C1被取消
* E--> D1被取消
* 最终输出E1
这里说明一点:在同步线程中,switchmap发射所有的Observables
在异步中,switchmap发射最近的Observables