RxJava3.0 操作符之转换操作符使用

转换操作符可用于对Observable发射的数据进行变换,最终观察者接收的是转换后我们需要的数据类型.本文主要介绍一些常用转换操作符的基本使用.

转换操作符分类

Map — 通过对序列的每一项都应用一个函数变换Observable发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项
FlatMap — 将原可观察对象Observable发射的每一条数据转换为单个可观察对象Observable，然后将这些Observable发射的数据平坦化的放进一个单独的Observable，可以认为是一个将嵌套的数据结构展开的过程。
GroupBy — 将原来的Observable拆分为一组Observable，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据
Buffer — 定期将可观察对象Observable中的项目收集到一个集合中并将他们打包发送，而不是一次发出一个项目.可以简单理解为先缓存一定数量项目,然后将他们一块发送
Scan — 对可观察对象Observable发出的每个项目按顺序应用一个函数，并发出每个连续值
Window — 定期将Observable的数据从可观察对象细拆分为可观察窗口并发出这些窗口，而不是一次发出一个项.

常见转换操作符的使用

Map

map 操作符

通过将函数应用于每个项目来转换可观察项发出的项

 //map 操作符private void mapOperator(){//原始数据 1,2,3 通过map 操作符,对原来每一项 进行乘10 最后转成String,最后观察者收到的就是转换后的每一条结果Observable.just(1,2,3).map(new Function<Integer, String>() {@Overridepublic String apply(Integer integer) throws Throwable {return integer*10 +"";}}).subscribe(new Consumer<String>() {@Overridepublic void accept(String s) throws Throwable {Log.i("sky>>>", "map : " + s);}});}输出:15:38:18.973 20278-20278/com.sky.rxjava I/sky>>>: map : 1015:38:18.973 20278-20278/com.sky.rxjava I/sky>>>: map : 2015:38:18.973 20278-20278/com.sky.rxjava I/sky>>>: map : 30

cast 操作符

cast操作符是 Map 的专用版本，它通过在重新发出源可观察项之前将其转换为特定类来转换源可观察项.相当于将原数据转换为指定class 类型数据,如果类型不匹配,会抛出ClassCastException异常.个人感觉此操作符有点鸡肋,可能个人水平有限,没理解这个运算符再实际开发中有什么用处,有大佬知道的,可以告知小弟.

    //cast 操作符private void castOperator() {Observable.just("hello", "RxJava").cast(String.class).subscribe(new Consumer<String>() {@Overridepublic void accept(String s) throws Throwable {Log.e("sky>>>", "cast : " + s);}});}//输出的还是原字符串.

`Flatmap`

flapMap/concatMap

flatMap 通过将指定的函数应用于源 Observable 发出的每个项目来转换可观察对象，其中该函数返回本身发出项目的可观察对象Observable 。然后，flatMap 合并这些生成的Observable 发射的数据，将这些合并的结果作为其自己的序列发出。

例如,当我们源Observable 发射的每一项就是一个可观察对象Observable 或者或者可以以其他方式转换为可观察对象时,这个操作符就很有用,可以创建一个新的Observable,此Observable 将发射这些次级Observable发射的数据的完整集合.

但是需要注意的是,新Observable 再合并这些子项Observable发射的数据时,不是有序的,数据可以交错.

先看一下最简单使用,源Observable 发射1,2,3,flatmap 操作符通过函数,将每一项转换换成一个ObservableSource返回回去,最后集合到一起,发射出去.

    private void flatMapOperator(){Observable.just(1,2,3).flatMap(new Function<Integer, ObservableSource<Integer>>() {@Overridepublic ObservableSource<Integer> apply(Integer integer) throws Throwable {List<Integer> list = new ArrayList<>();for (int i = 0; i < integer; i++) {list.add(integer*10 +i);}return Observable.fromIterable(list);}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Throwable {Log.i("sky>>>", "flatMap: " + integer);}});}输出:11530-11530/com.sky.rxjava I/sky>>>: flatMap: 1011530-11530/com.sky.rxjava I/sky>>>: flatMap: 2011530-11530/com.sky.rxjava I/sky>>>: flatMap: 2111530-11530/com.sky.rxjava I/sky>>>: flatMap: 3011530-11530/com.sky.rxjava I/sky>>>: flatMap: 3111530-11530/com.sky.rxjava I/sky>>>: flatMap: 32

之前有提到 flatMap 合并数据时是无序的,可能会出现数据交错的情况,但此处打印,显示数据是正常的.

原因猜测:

1.数据量过小

2.此时并未做线程切换,或者耗时操作.

改进测试代码,增加数据量,子Observable增加随机延迟,增加线程切换,

    private void flatMapOperator(){//增加数据量Observable.just(10,20,30).flatMap(new Function<Integer, ObservableSource<Integer>>() {@Overridepublic ObservableSource<Integer> apply(Integer integer) throws Throwable {List<Integer> list = new ArrayList<>();for (int i = 0; i < integer; i++) {list.add(integer*10 +i);}//加了随机延迟return Observable.fromIterable(list).delay(new Random().nextInt(5), TimeUnit.SECONDS);}})       //增加了线程的切换.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Throwable {Log.i("sky>>>", "flatMap: " + integer);}});}
//由于输出太多,只截取一部分看看结果,此时看到,最终收到数据的顺序已不再有序,发生穿插出现的情况
输出:
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 106
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 107
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 108
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 109
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 200
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 300
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 301
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 201
20231-20231/com.sky.rxjava I/sky>>>: flatMap: 302

如果我们想要的结果就是有序的怎么不处理,这时可使用另一个操作 concatMap ,concatMap 操作符与flatMap 操作符用法一模一样,区别就是 flatMap合并后的数据无序,而concatMap 最终的数据结果也是有序的.

由于flatMap 还是一个比较常用的操作符,再写个例子

假设有一个学生类,每个学生都有自己报名的兴趣班,我们的源Observable 发射的是学生对象,而我们观察者需要的结果是输出每个学生所报的兴趣班,这时我们就可以使用flatMap 去处理.

    /*** 定义学生类*/public  class Student{public String name;public List<String> subjects;/**** @param name 学生姓名* @param subjects 所报的科目*/public Student(String name,String ... subjects) {this.name = name;this.subjects = new ArrayList<>();for (String subject : subjects) {this.subjects.add(name +": "+ subject);}}}private void flatMapOperator2(){Student s1 = new Student("刘德华", "武术", "表演", "唱歌");Student s2 = new Student("周杰伦", "赛车", "钢琴");Student s3 = new Student("菜虚鲲", "唱", "跳", "rap", "篮球");Observable.just(s1,s2,s3).flatMap(new Function<Student , ObservableSource<String>>() {@Overridepublic ObservableSource<String> apply(Student student) throws Throwable {return Observable.fromIterable(student.subjects);}}).subscribe(new Consumer<String>() {@Overridepublic void accept(String subject) throws Throwable {Log.i("sky>>>", subject );}});}输出:25298-25298/com.sky.rxjava I/sky>>>: 刘德华: 武术25298-25298/com.sky.rxjava I/sky>>>: 刘德华: 表演25298-25298/com.sky.rxjava I/sky>>>: 刘德华: 唱歌25298-25298/com.sky.rxjava I/sky>>>: 周杰伦: 赛车25298-25298/com.sky.rxjava I/sky>>>: 周杰伦: 钢琴25298-25298/com.sky.rxjava I/sky>>>: 菜虚鲲: 唱25298-25298/com.sky.rxjava I/sky>>>: 菜虚鲲: 跳25298-25298/com.sky.rxjava I/sky>>>: 菜虚鲲: rap25298-25298/com.sky.rxjava I/sky>>>: 菜虚鲲: 篮球

`map` 与`flatMap`区别

入参

//map
public final <R> Observable<R> map(@NonNull Function<? super T, ? extends R> mapper) {Objects.requireNonNull(mapper, "mapper is null");return RxJavaPlugins.onAssembly(new ObservableMap<>(this, mapper));}//flatMap
public final <R> Observable<R> flatMap(@NonNull Function<? super T, ? extends ObservableSource<? extends R>> mapper) {return flatMap(mapper, false);
}@FunctionalInterface
public interface Function<@NonNull T, @NonNull R> {R apply(T t) throws Throwable;
}

从两个函数定义来看,两个操作符入参都是一个Function 的函数接口,并返回一个Observable.

map操作符入参函数接口的第二个泛型参数可以为任意类型.

flatMap 操作符入参函数接口第二个泛型参数必须是一个ObservableSource 类型

转换返回

操作符传递进来的函数接口第二个类型,就是要对每一项进行转换后返回的类型

map 操作符对数据变换可以返回任意类型的值.并且与源数据一一对应,所以说map操作符是转换数据是 1对1的关系.

flatMap 操作符数据变换转换后必须返回ObservableSource 类型,最终将这些返回的ObservableSource 发射项合并，将这些合并的结果作为其自己的序列发出.由于转换源Observable 每一项转换后是一个ObservableSource 类型,也就是说,他也是一个可发送一个或多个Observable.源Observable 的每一项可被拆分成多个,所以说 map操作符是一对多/多对多的变换.

`GroupBy`

groupBy操作符将原始Observable分拆为一些Observable集合，通过Function函数接口,为原数据每一项指定一个 key ,相同的 key 数据会被同一个Observable发射.

例, 源Observable 发出一组姓名,使用groupBy 根据他们的姓分为了三组,周姓一组,李姓一组,其他一组,分别给他们指定了 key “zhou”,“li”,“other”

groupBy 返回了Observable<GroupedObservable<K, T>>, 一个数据类型为GroupedObservable<K,T>的可观察者.

GroupedObservable又是一个带有K 类型key,发射数据类型为T 的Observable,在我们这个例子里,这个返回的可观察者会发送三项 GroupedObservable<String,String>数据,将周姓,李姓,其他姓分别归为一组,同一个key的数据将由同一个GroupedObservable<String, String>去发送.通过其getKey()函数可以获取到当前Observable的key值.

由日志可以看出, “收到 key” 的日志打印了三次,说明源Observable发送的数据确实是根据我们指定的key 被分成了三个Observable,然后每个Observable分别发送各自的数据.

    private void groupByOperator() {Observable.just("周杰伦", "李连杰", "孙燕姿", "周星驰", "李荣浩", "周润发", "李宇春", "林俊杰").groupBy(new Function<String, String>() {@Overridepublic String apply(String s) throws Throwable {if (s.startsWith("周")) {return "zhou";} else if (s.startsWith("李")) {return "li";}return "other";}}).subscribe(new Consumer<GroupedObservable<String, String>>() {@Overridepublic void accept(GroupedObservable<String, String> stringStringGroupedObservable) throws Throwable {Log.i("sky>>>", "收到 key 为 " + stringStringGroupedObservable.getKey() + " 的一组数据的Observable");stringStringGroupedObservable.subscribe(new Consumer<String>() {@Overridepublic void accept(String s) throws Throwable {String key = stringStringGroupedObservable.getKey();Log.i("sky>>>", "groupBy : key= " + key + " , value = " + s);}});}});}输出: 18759-18759/com.sky.rxjava I/sky>>>: 收到  key 为 zhou 的一组数据的Observable       //收到key 为 zhou的Observable18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= zhou , value = 周杰伦18759-18759/com.sky.rxjava I/sky>>>: 收到  key 为 li 的一组数据的Observable         //收到key 为li的Observable18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= li , value = 李连杰18759-18759/com.sky.rxjava I/sky>>>: 收到  key 为 other 的一组数据的Observable      //收到key 为other的Observable18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= other , value = 孙燕姿18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= zhou , value = 周星驰18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= li , value = 李荣浩18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= zhou , value = 周润发18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= li , value = 李宇春18759-18759/com.sky.rxjava I/sky>>>: groupBy : key= other , value = 林俊杰

`Buffer`

定期将可观察对象Observable中的项目收集到一个集合中并将他们打包发送，而不是一次发出一个项目.可以简单理解为先缓存一定数量项目,然后将他们一块发送

    private void bufferOperator(){//buffer(int count)Observable.just("a","b","c","d","e","f").buffer(2).subscribe(new Consumer<List<String>>() {@Overridepublic void accept(List<String> strings) throws Throwable {for (String string : strings) {Log.i("sky>>>", "buffer: " + string);}Log.w("sky>>>", "-------------");}});输出:28696-28696/com.sky.rxjava I/sky>>>: buffer: a28696-28696/com.sky.rxjava I/sky>>>: buffer: b28696-28696/com.sky.rxjava W/sky>>>: -------------28696-28696/com.sky.rxjava I/sky>>>: buffer: c28696-28696/com.sky.rxjava I/sky>>>: buffer: d28696-28696/com.sky.rxjava W/sky>>>: -------------28696-28696/com.sky.rxjava I/sky>>>: buffer: e28696-28696/com.sky.rxjava I/sky>>>: buffer: f28696-28696/com.sky.rxjava W/sky>>>: -------------   //buffer(int count,int skip)Observable.just(0,1,2,3,4,5,6,7,8,9).buffer(2,3).subscribe(new Consumer<List<Integer>>() {@Overridepublic void accept(List<Integer> integers) throws Throwable {for (Integer i : integers) {Log.i("sky>>>", "buffer skip: " + i);}Log.w("sky>>>", "-------------");}});}输出:28696-28696/com.sky.rxjava I/sky>>>: buffer skip: 028696-28696/com.sky.rxjava I/sky>>>: buffer skip: 128696-28696/com.sky.rxjava W/sky>>>: -------------28696-28696/com.sky.rxjava I/sky>>>: buffer skip: 328696-28696/com.sky.rxjava I/sky>>>: buffer skip: 428696-28696/com.sky.rxjava W/sky>>>: -------------28696-28696/com.sky.rxjava I/sky>>>: buffer skip: 628696-28696/com.sky.rxjava I/sky>>>: buffer skip: 728696-28696/com.sky.rxjava W/sky>>>: -------------28696-28696/com.sky.rxjava I/sky>>>: buffer skip: 928696-28696/com.sky.rxjava W/sky>>>: -------------

第一个例子中,buffer(2) 内部实际调用了 buffer(2,2), 与第二个例子一致, count 表示每个缓冲区要存储几条数据, skip表示在启动新缓冲区之前，应跳过当前可观察对象发出的项目数.

当 count == skip 时,我们第一个例子.就是每次两个数据为一组放置到集合中,并且新缓冲区刚好跳过skip (等于 count )条数据,所以数据就以 count 条为一组分组放置到集合中,最后由Observable发送这些集合.(不会存在数据跳过或者重叠现象).

当 count < skip 时,第二个例子,将 count 条数据放置在集合中,新缓冲区要跳过 skip 条数据,由于 skip 大于 count ,多余出来的数据就会被丢弃,不会再新缓冲区集合存放,最后也是由新创建的 Observable 将这些结集合发射出去.(会存在数据跳过)

当 count > skip 时,将 count 条数据放置在缓冲集合中,由于 skip 小于 count ,再启动新缓冲时,调过 skip 条数据后,会存在数据重叠的现象.

buffer还有其他多个重载函数,但由于实际用处不多,这里就不在多做介绍.

`Scan`

scan 运算符将函数应用于源Observable发出的第一个数据，然后将该函数的结果作为其自己的第一项数据发射。它还将函数的结果与源Observable发出的第二个项目一起反馈到函数中，以生成其第二个发射数据。它持续将每次结果与下一个数据项反馈函数中，以创建其序列的其余部分

这个操作符在某些情况下被叫做accumulator(累加器)。

例我们源Observable 发射 “h”,“e”,“l”,“l”,“o”,“!” 这些字符,由于第一项数据只有他自身,所以没有调用apply去生成第一条发送项,而是将他自身作为第一条发送项的结果,传递进去,与第二项进行累加生成后续的输出.相当于一直再一层一层的包装累加,直到最后一项.

    private void scanOperator(){Observable.just("h","e","l","l","o","!").scan(new BiFunction<String, String, String>() {@Overridepublic String apply(String s, String s2) throws Throwable {Log.i("sky>>>","s =  "+ s+ " s2 = "+s2);return s+ s2;}}).subscribe(new Consumer<String>() {@Overridepublic void accept(String s) throws Throwable {Log.i("sky>>>","scan : "+ s);}});}输出:3428-3428/com.sky.rxjava I/sky>>>: scan : h3428-3428/com.sky.rxjava I/sky>>>: s =  h s2 = e3428-3428/com.sky.rxjava I/sky>>>: scan : he3428-3428/com.sky.rxjava I/sky>>>: s =  he s2 = l3428-3428/com.sky.rxjava I/sky>>>: scan : hel3428-3428/com.sky.rxjava I/sky>>>: s =  hel s2 = l3428-3428/com.sky.rxjava I/sky>>>: scan : hell3428-3428/com.sky.rxjava I/sky>>>: s =  hell s2 = o3428-3428/com.sky.rxjava I/sky>>>: scan : hello3428-3428/com.sky.rxjava I/sky>>>: s =  hello s2 = !3428-3428/com.sky.rxjava I/sky>>>: scan : hello!

`window`

定期将项目从可观察对象细分为可观察窗口(Observable)并发出这些窗口，而不是一次发出一个项目

window类似于buffer，但它不是从源Observable发出项目数据集合包，而是发出可观察对象，每个可观察对象从源可观察对象发出一部分项目，然后终止并发出最后发射一个onCompleted通知.

例: 源数据发送1到8,window 按照每两项创建一个新Observable,最后将这些新创建的Observable作为可观察对象数据项发射出去.最终观察者接收到的还是一个可观察对象,需要重新订阅,以接收到每个可观察对象的的数据项.

window 也有很多重载函数,这里也不再一一介绍了.

    private void windowOperator(){Observable.just(1,2,3,4,5,6,7,8).window(2).subscribe(new Consumer<Observable<Integer>>() {@Overridepublic void accept(Observable<Integer> integerObservable) throws Throwable {Log.i("sky>>>","-------------");integerObservable.subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Throwable {Log.i("sky>>>","window : "+ integer);}});}});}输出 :10390-10390/com.sky.rxjava I/sky>>>: -------------10390-10390/com.sky.rxjava I/sky>>>: window : 110390-10390/com.sky.rxjava I/sky>>>: window : 210390-10390/com.sky.rxjava I/sky>>>: -------------10390-10390/com.sky.rxjava I/sky>>>: window : 310390-10390/com.sky.rxjava I/sky>>>: window : 410390-10390/com.sky.rxjava I/sky>>>: -------------10390-10390/com.sky.rxjava I/sky>>>: window : 510390-10390/com.sky.rxjava I/sky>>>: window : 610390-10390/com.sky.rxjava I/sky>>>: -------------10390-10390/com.sky.rxjava I/sky>>>: window : 710390-10390/com.sky.rxjava I/sky>>>: window : 8

end

RxJava常见转换操作符总结完毕