RxJava学习之转换型操作符
标签（空格分隔）： RX系列
转换型操作符
下面展示了可用于Observable发射的数据执行变换操作的各种操作符
- map()---对序列的每一项都应用一个函数来变换Observable发射的数据序列
- flatMap()、concatMap()、flatMapIterable()---将Observable发射的数据集合变换为Observables集合,然后将这些Observable发射的数据平坦化的放进一个单独的Observable
- switchMap()---将Observable发射的数据集合变换为Observables集合,然后只发射这些Observables最近发射的数据
- scan()---对Observable发射的每一项数据应用一个函数,然后按照顺序依次发射每一个值
- groupyBy()---将Observable分拆为Observable集合,将原始的Observable发射的数据按照key分组,每一个Observable发射一组不同的数据
- buffer()---它顶起从Observable收集数据到一个集合,然后把这些数据集合打包发射,而不是一次发射
- window()---定期将来自Observable的数据分拆成一些Observable窗口,然后发射这些窗口,而不是每次发射一项
- cast()---在发射之前强制将Observable发射的所有哦数据转换为指定数据类型
map操作符
对Observable发射的每一项数据应用一个函数，执行变换操作
Map操作符对原始Observable发射的每一项数据应用一个你选择的函数，然后返回一个发射这些结果的Observable。
RxJava将这个操作符实现为map函数。这个操作符默认不在任何特定的调度器上执行。
Map操作符的源码
/**
* Returns an Observable that applies a specified function to each item emitted by the source Observable and
* emits the results of these function applications.
* <p>
* 
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code map} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param func
* a function to apply to each item emitted by the Observable
* @return an Observable that emits the items from the source Observable, transformed by the specified
* function
* @see <a href="http://reactivex.io/documentation/operators/map.html">ReactiveX operators documentation: Map</a>
*/
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
return lift(new OperatorMap<T, R>(func));
}
//可以看到接受一个Func1函数
/**
* Represents a function with one argument.
* @param <T> the first argument type 作为参数类型
* @param <R> the result type 作为返回参数类型
*/
public interface Func1<T, R> extends Function {
R call(T t);
}
map例子
static List<Student> studentList = new ArrayList<Student>(){
{
add(new Student("FAALLDA", 28));
add(new Student("小弟弟", 23));
add(new Student("妻子的孜孜", 25));
}
};
/**
* Map
*
* 通过使用map中的方法对Observable中发射出来的所有数据进行变换
*
* test1()方法是得到多个Student对象中的name，保存到nameList中
* 注意：接口Func1包装的是有返回值的方法。
*
*/
private static void test1(){
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 onNext(String value) {
System.out.println("onSuccess value = " + value);
nameList.add(value);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
}
Map操作符连续使用
/**
* Map操作符连续使用
*/
private static void test(){
Observable.from(studentList)
//把student转换成Integer
.map(new Func1<Student, Integer>() {
@Override
public Integer call(Student student) {
return student.age;
}
})
//将Integer转换成String
.map(new Func1<Integer, String>() {
@Override
public String call(Integer t) {
return String.valueOf(t+10);
}
})
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted ");
}
@Override
public void onNext(String value) {
System.out.println("onSuccess value = " + value);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
}
Flatmap操作符
FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable
FlatMap操作符使用一个指定的函数对原始Observable发射的每一项数据执行变换操作，这个函数返回一个本身也发射数据的Observable，然后FlatMap合并这些Observables发射的数据，最后将合并后的结果当做它自己的数据序列发射。
这个方法是很有用的，例如，当你有一个这样的Observable：它发射一个数据序列，这些数据本身包含Observable成员或者可以变换为Observable，因此你可以创建一个新的Observable发射这些次级Observable发射的数据的完整集合。
注意：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。
/**
* FlatMap操作符
* FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable
*/
private static void test4(){
List<String> nameList = new ArrayList<>();
Observable.from(studentList)
//第一个Student类型使我们的flatMap返回函数call里面的参数类型
//第二个Entity类型使我们的flatMap中的call的返回值类型
.flatMap(new Func1<Student, Observable<Entity>>() {
@Override
public Observable<Entity> call(Student student) {
Course course = couseMap.get(student.name);
Entity entity = new Entity(course, student);
return Observable.just(entity);
}
})
.subscribe(new Subscriber<Entity>() {
@Override
public void onCompleted() {
}
@Override
public void onNext(Entity entity) {
System.out.println("onSuccess entity = " + entity);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
}
concatMap操作符
还有一个concatMap操作符，它类似于最简单版本的flatMap，但是它按次序连接而不是合并那些生成的Observables，然后产生自己的数据序列。
/**
* ConcatMap操作符
* 类似于最简单版本的flatMap，但是它按次序连接而不是合并那些生成的Observables，然后产生自己的数据序列。
*
*/
private static void test5(){
Observable.from(studentList)
.concatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student t) {
Course course = couseMap.get(t.name);
return Observable.just(course);
}
})
.subscribe(new Subscriber<Course>() {
@Override
public void onCompleted() {
}
@Override
public void onNext(Course course) {
System.out.println("onSuccess course = " + course);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
}
concatMap和flatMap的区别
/**
* flatMap与ConcatMap操作符比较
* 区别：
* 无序：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。
* 有序：ConcatMap不会让变换后的Observables发射的数据交错，它按照严格的顺序发射这些数据。
*
* 说明：在同步线程中，FlatMap和ConcactMap的执行结果是一样的（结果是有序的），
* 只有在异步线程中，FlatMap结果可能是无序的，而ConcactMap始终能保持有序的结果。
*
* concatMap与flatMap操作符的比较
*/
private static void test(){
List<Integer> numbers = Arrays.asList(2, 3, 4, 5, 6, 7, 8, 9, 10);
Observable.from(numbers)
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer t) {
return Observable.just(t).subscribeOn(Schedulers.from(Executors.newCachedThreadPool()));
//return Observable.just(t);
}
})
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onNext(Integer value) {
System.out.println("flatMap onSuccess value = " + value);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
System.out.println("----------------------------");
Observable.from(numbers)
.concatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer t) {
return Observable.just(t).subscribeOn(Schedulers.from(Executors.newCachedThreadPool()));
//return Observable.just(t);
}
})
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onNext(Integer value) {
System.out.println("concatMap onNext value = " + value);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
}
switchMap操作符
- 它和flatMap很像，除了一点：当原始Observable发射一个新的数据（Observable）时，它将取消订阅并停止监视产生执之前那个数据的Observable，只监视当前这一个
/**
* switchMap
* 解释：将Observable发射的数据集合变换为Observables集合，然后只发射这些Observables最近发射的数据
* 用法与FlatMap几乎一样，区别是SwitchMap操作符只会发射[emit]最近的Observables。
*
* 当源Observable发射一个新的数据项时，如果旧数据项订阅还未完成，就取消旧订阅数据和停止监视那个数据项产生的Observable,开始监视新的数据项.
*
* 应用场景：http://blog.csdn.net/jdsjlzx/article/details/51730162
*
* 逻辑推演：
* A --> 取消空的，没有可以取消的
* B--> A1被取消
* C--> B1被取消
* D--> C1被取消
* E--> D1被取消
* 最终输出E1
*/
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();
}
}
switchMap与flatmap的区别
/**
* switchMap与flatmap的区别
*
* 说明：在同步线程中，switchMap发射[emit]所有的Observables，
* 在异步线程中，switchMap只会发射[emit]最近的Observables。
*
*/
private static void test8(){
ExecutorService service = Executors.newFixedThreadPool(10);
List<Integer> numbers = Arrays.asList(2, 3, 4, 5, 6, 7, 8, 9, 10);
Observable.from(numbers)
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer t) {
return Observable.just(t).subscribeOn(Schedulers.from(service));
//return Observable.just(t);
}
})
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onNext(Integer value) {
System.out.println("flatMap onNext value = " + value);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
System.out.println("----------------------------------");
Observable.from(numbers)
.switchMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer t) {
return Observable.just(t).subscribeOn(Schedulers.from(service));
//return Observable.just(t);
}
})
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onNext(Integer value) {
System.out.println("switchMap2 onNext value = " + value);
}
@Override
public void onError(Throwable error) {
System.out.println("onError error = " + error);
}
});
service.shutdown();
}
