RxJava2系列之转换型操作符

原文首发于微信公众号：躬行之(jzman-blog)，欢迎关注交流！

1. buffer 操作符
2. window操作符
3. map操作符
4. groupBy操作符
5. cast操作符
6. scan操作符
7. To操作符

buffer 操作符

buffer 操作符重载方法比较多，这里选取典型的几个来说明 buffer 操作符的使用，buffer 操作符的使用可以分为如下三类，具体如下：

//第一类
public final Observable<List<T>> buffer(int count) 
public final Observable<List<T>> buffer(int count, int skip) 
//第二类
public final Observable<List<T>> buffer(long timespan, TimeUnit unit)
public final Observable<List<T>> buffer(long timespan, long timeskip, TimeUnit unit) 
//第三类
public final <B> Observable<List<T>> buffer(ObservableSource<B> boundary)
public final <TOpening, TClosing> Observable<List<T>> buffer(
            ObservableSource<? extends TOpening> openingIndicator,
            Function<? super TOpening, ? extends ObservableSource<? extends TClosing>> closingIndicator)
复制代码

buffer(int count)

buffer 操作符将一个 Observable 转换为一个 Observable,这个 Observable 用于收集原来发送的数据，然后发送这些缓存的数据集合,buffer 将发送的单个事件转换成元素集合,下面是针对此种情况的官方示意图：

如下面的事件的发送过程，如果不设置 buffer 则需要发送四次，如果使用如下 buffer 进行转换，则只需发送两次,测试代码如下：

count = 0;
Observable.just("Event1", "Event2", "Event3", "Event4")
        .buffer(2)
        .subscribe(new Consumer<List<String>>() {
            @Override
            public void accept(List<String> strings) throws Exception {
                count++;
                Log.i(TAG, "第" + count + "次接收...");
                Log.i(TAG, "accept--->" + strings.size());
                Log.i(TAG, "接收的数据...");
                for (String str : strings) {
                    Log.i(TAG, "accept--->" + strings.size() + "---" + str);
                }
            }
        });
复制代码

上述代码的执行结果如下：

第1次接收...
accept--->2
接收的数据...
accept--->2---Event1
accept--->2---Event2
第2次接收...
accept--->2
接收的数据...
accept--->2---Event3
accept--->2---Event4
复制代码

buffer(int count, int skip)

相较 buffer(int count), skip 可以指定下一次由源 Observable 转换的 Observable 收集事件的位置，如果 count 等于 skip,则 buffer(int count,int skip) 等价于 buffer(int count)，官方示意图如下：

如下面的事件发送过程，相当于每 3 个事件一组进行发送，但每次收集数据的位置参数 skip 为 2，则每次收集的数据中会有数据重复，测试代码如下：

count = 0;
Observable.just("Event1", "Event2", "Event3", "Event4", "Event5")
        .buffer(3, 2)
        .subscribe(new Consumer<List<String>>() {
            @Override
            public void accept(List<String> strings) throws Exception {
                count++;
                Log.i(TAG, "第" + count + "次接收...");
                Log.i(TAG, "accept--->" + strings.size());
                Log.i(TAG, "接收的数据...");
                for (String str : strings) {
                    Log.i(TAG, "accept--->" + strings.size() + "---" + str);
                }
            }
        });
复制代码

上述代码的执行结果如下：

第1次接收...
accept--->3
接收的数据...
accept--->3---Event1
accept--->3---Event2
accept--->3---Event3
第2次接收...
accept--->3
接收的数据...
accept--->3---Event3
accept--->3---Event4
accept--->3---Event5
第3次接收...
accept--->1
接收的数据...
accept--->1---Event5
复制代码

buffer(long timespan, TimeUnit unit)

buffer 操作符会将一个 Observable 转换为一个新的 Observable,timespan 决定新的的 Observsable 在发出缓存的数据的时间间隔，官方示意图如下：

如下面的事件发送过程，源 Observable 每隔 2 秒发送事件，而 buffer 新生成的 Obsrevable 则以每隔 1 秒的间隔发送缓存的事件集合，当然，这样会在间隔的时间段收集不到数据导致丢失数据，测试代码如下：

Observable.intervalRange(1,8,0,2, TimeUnit.SECONDS)
        .buffer(1,TimeUnit.SECONDS)
        .subscribe(new Consumer<List<Long>>() {
            @Override
            public void accept(List<Long> longs) throws Exception {
                Log.i(TAG, "accept--->" + String.valueOf(longs));
            }
        });
复制代码

上述代码的执行结果如下：

accept--->[1]
accept--->[]
accept--->[2]
accept--->[]
accept--->[3]
accept--->[]
accept--->[4]
accept--->[]
accept--->[5]
复制代码

buffer(long timespan, long timeskip, TimeUnit unit)

buffer 操作符会将一个 Observable 转换为一个 Observable,timeskip 决定让新生成的 Observable 定期启动一个新的缓冲区，然后新的 Observable 会发出在 timespan 时间间隔内收集的事件集合，官方示意图如下：

如下面的事件发送过程，源 Observable 会每隔 1 秒发送 1 到 12 的整数，buffer 新生成的 Observable 会每隔 5 秒接收源 Observable 发送的事件，测试代码如下：

Observable.intervalRange(1,12,0,1, TimeUnit.SECONDS)
        .buffer(1,5, TimeUnit.SECONDS)
        .subscribe(new Consumer<List<Long>>() {
            @Override
            public void accept(List<Long> longs) throws Exception {
                Log.i(TAG, "accept--->" + String.valueOf(longs));
            }
        });
复制代码

上述代码的执行结果如下：

accept--->[1]
accept--->[6]
accept--->[11]
复制代码

buffer(ObservableSource boundary)

buffer(boundary) 会监视一个名叫 boundary 的 Observable，每当这个 Observable 发射了一个事件，它就创建一个新的 List 开始收集来自原始 Observable 的发送的事件并发送收集到的数据,官方示意图如下：

如下面事件的发送过程，收集到的原事件会因为时间间隔的不同最终发送的收集到的事件的个数也不同，测试代码如下：

Observable.intervalRange(1,10,0,2, TimeUnit.SECONDS)
        .buffer(Observable.interval(3, TimeUnit.SECONDS))
        .subscribe(new Consumer<List<Long>>() {
            @Override
            public void accept(List<Long> longs) throws Exception {
                Log.i(TAG, "accept--->" + String.valueOf(longs));
            }
        });
复制代码

上述代码的执行结果如下：

accept--->[1, 2]
accept--->[3]
accept--->[4, 5]
accept--->[6]
accept--->[7, 8]
accept--->[9]
accept--->[10]
复制代码

buffer(openingIndicator, closingIndicator)

buffer(openingIndicator, closingIndicator)会监视一个名叫 openingIndicator 的 Observable，这个 Observable 每发射一个事件，它就创建一个 List 收集原始 Observable 发送的数据，并将收集的数据给 closingIndicator，closingIndicator 会返回一个 Observable,这个 buffer 会监视 closingIndicator 返回的Observable,检测到这个 Observable 的数据时，就会关闭这个 List 发射刚才从 openingIndicator 获得数据，也就是名为 openingIndicator 的 Observable 收集的数据，下面是针对此种情况的官方示意图：

如下面时间发送过程，原始的 Observable 以每个 1 秒的间隔发送 1 到 12 之间的整数，名为 openingIndicator 的 Observable 会每隔 3 秒创建一个 List 手机发送的事件，然后将收集的数据给 closingIndicator，closingIndicator 会延时 1 秒发送从名为 openingIndicator 的 Observable 拿到的数据，下面是测试代码：

Observable openingIndicator = Observable.interval(3, TimeUnit.SECONDS);
Observable closingIndicator = Observable.timer(1,TimeUnit.SECONDS);
Observable.intervalRange(1,12,0,1, TimeUnit.SECONDS)
        .buffer(openingIndicator, new Function<Object, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Object o) throws Exception {
                return closingIndicator;
            }
        })
        .subscribe(new Consumer<List<Long>>() {
            @Override
            public void accept(List<Long> longs) throws Exception {
                Log.i(TAG, "accept--->" + String.valueOf(longs));
            }
        });
复制代码

上述代码的执行结果如下：

accept--->[4, 5]
accept--->[7]
accept--->[10]
复制代码

window操作符

这里就以 window(long count) 为例来介绍 window 操作符的使用，window 操作符的使用和 buffer 使用类似，不同之处是经 buffer 转换成的 Observable 发送的时源 Observable 发送事件的事件集合，而经 window 操作符转换成的 Observable 会依次发送 count 个源 Observable 发送的事件，该操作符官方示意图如下：

测试代码如下：

Observable.just("Event1", "Event2", "Event3", "Event4")
        .window(2)
        .subscribe(new Consumer<Observable<String>>() {
            @Override
            public void accept(Observable<String> stringObservable) throws Exception {
                Log.i(TAG, "accept--Observable->");
                stringObservable.subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG, "accept--->" + s);
                    }
                });
            }
        });
复制代码

上述代码的执行结果如下：

accept--Observable->
accept--->Event1
accept--->Event2
accept--Observable->
accept--->Event3
accept--->Event4
复制代码

map操作符

map(mapper)

map 操作符可对发送的数据进行相应的类型转化,map 操作的官方示意图如下：

如下面的事件发送过程，经过 map 操作符转换，可对源 Observable 发送的事件进行进一步的加工和转换，测试代码如下：

Observable.just("Event1", "Event2", "Event3", "Event4")
        .map(new Function<String, String>() {
            @Override
            public String apply(String s) throws Exception {
                return "this is " + s;
            }
        }).subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        Log.i(TAG, "accept--->" + s);
    }
});
复制代码

上述代码的执行结果如下：

accept--->this is Event1
accept--->this is Event2
accept--->this is Event3
accept--->this is Event4
复制代码

flatMap(mapper)

flatMap 操作符使用时，当源 Observable 发出事件会相应的转换为可以发送多个事件的 Observable，这些 Observable 最终汇入同一个 Observable，然后这个 Observable 将这些事件统一发送出去，这里决定不再想上文中一样，每个重载方法都进行说明，这里已常用的 flatMap(mapper) 为例，其官方示意图如下：

如下面的事件发送过程，使用了 flatMap 操作符之后，源 Observable 发送事件时，相应的生成对应的 Observable，最终发送的事件都汇入同一个 Observable,然后将事件结果回调给观察者，测试代码如下：

final Observable observable = Observable.just("Event5", "Event6");
Observable.just("Event1", "Event2", "Event3", "Event4")
        .flatMap(new Function<String, Observable<String>>() {
            @Override
            public Observable<String> apply(String s) throws Exception {
                return observable;
            }
        }).subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        Log.i(TAG, "accept--->" + s);
    }
});
复制代码

上述代码的执行结果如下：

accept--->Event5
accept--->Event6
accept--->Event5
accept--->Event6
accept--->Event5
accept--->Event6
accept--->Event5
accept--->Event6
复制代码

concatMap(mapper)

concatMap 的使用与 flatMap 的使用大致类似，相较flatMap能够保证事件接收的顺序，而 flatMap 不能保证事件接收的顺序，concatMap 操作符的官方示意图如下：

如下面的事件发送过程，我们在源 Observable 发送整数 1 时延时 3 秒，然后继续发送其他事件，下面是测试代码：

Observable.intervalRange(1, 2, 0, 1, TimeUnit.SECONDS)
        .concatMap(new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) throws Exception {
                int delay = 0;
                if (aLong == 1) {
                    delay = 3;
                }
                return Observable.intervalRange(4, 4, delay, 1, TimeUnit.SECONDS);
            }
        }).subscribe(new Consumer<Long>() {
    @Override
    public void accept(Long aLong) throws Exception {
        Log.i(TAG, "accept--->" + aLong);
    }
});

复制代码

使用 concatMap 操作符上述代码的执行结果如下：

accept--->4
accept--->5
accept--->6
accept--->7
accept--->4
accept--->5
accept--->6
accept--->7
复制代码

使用 flatMap 操作符上述代码的执行结果如下：

accept--->4
accept--->5
accept--->6
accept--->4
accept--->7
accept--->5
accept--->6
accept--->7
复制代码

可见，concatMap 相较 flatMap 能够保证事件接收的顺序。

switchMap(mapper)

当源 Observable 发送事件时会相应的转换为可以发送多个事件的 Observable，switchMap 操作符只关心当前这个 Observable，也就是说，源 Observable 每当发送一个新的事件时，就会丢弃前面一个发送多个事件的 Observable，官方示意图如下：

如下面的事件发送过程，源 Observable 每个 2 秒发送 1 和 2，转换成的可以发送多个事件的 Observable 每个 1 秒发送从 4 开始的整数，使用 switchMap 操作符时，源 Observable 发送一个整数 1 时，这个新的可以发送多个事件的 Observable 只发送两个整数，也就是 4 和 5 之后就停止发送了，因为此时源 Observable 又开始发送事件了，此时会丢弃前一个可发送多个时间的 Observable,开始下一次源 Observable 发送事件的监听，测试代码如下：

Observable.intervalRange(1, 2, 0, 2, TimeUnit.SECONDS)
        .switchMap(new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) throws Exception {
                Log.i(TAG, "accept-aLong-->" + aLong);
                return Observable.intervalRange(4, 4, 0, 1, TimeUnit.SECONDS);
            }
        }).subscribe(new Consumer<Long>() {
    @Override
    public void accept(Long aLong) throws Exception {
        Log.i(TAG, "accept--->" + aLong);
    }
});
复制代码

上述代码执行结果如下：

accept-aLong-->1
accept--->4
accept--->5
accept-aLong-->2
accept--->4
accept--->5
accept--->6
accept--->7
复制代码

此外，还有与之相关的操作符：concatMapDelayError、concatMapEager、concatMapEagerDelayError concatMapIterable、flatMapIterable 、switchMapDelayError，都是上述操作符的扩展，这里就不在介绍了。

groupBy操作符

groupBy 操作符会对接收的数据按照指定的规则进行分类，然后再被 GroupedObservable 等订阅输出，官方示意图如下：

如下面的事件发送过程，我们会按照成绩进行分组输出，具体如下：

List<DataBean> beanList = new ArrayList<>();
beanList.add(new DataBean("成绩是95分", 95));
beanList.add(new DataBean("成绩是70分", 70));
beanList.add(new DataBean("成绩是56分", 56));
beanList.add(new DataBean("成绩是69分", 69));
beanList.add(new DataBean("成绩是90分", 90));
beanList.add(new DataBean("成绩是46分", 46));
beanList.add(new DataBean("成绩是85分", 85));

Observable.fromIterable(beanList)
        .groupBy(new Function<DataBean, String>() {
            @Override
            public String apply(DataBean dataBean) throws Exception {
                int score = dataBean.getScore();
                if (score >= 80) {
                    return "A";
                }

                if (score >= 60 && score < 80) {
                    return "B";
                }

                if (score < 60) {
                    return "C";
                }
                return null;
            }
        })
        .subscribe(new Consumer<GroupedObservable<String, DataBean>>() {
            @Override
            public void accept(final GroupedObservable<String, DataBean> groupedObservable) throws Exception {
                groupedObservable.subscribe(new Consumer<DataBean>() {
                    @Override
                    public void accept(DataBean dataBean) throws Exception {
                        Log.i(TAG, "accept--->"+ groupedObservable.getKey() + "组--->"+dataBean.getDesc());
                    }
                });
            }
        });

复制代码

上述代码的执行结果如下：

accept--->A组--->成绩是95分 
accept--->B组--->成绩是70分
accept--->C组--->成绩是56分
accept--->B组--->成绩是69分
accept--->A组--->成绩是90分
accept--->C组--->成绩是46分
accept--->A组--->成绩是85分
复制代码

cast操作符

cast 操作符用于类型转化，cast 操作符官方示意图如下：

测试代码如下：

Observable.just(1,2,3,4,5)
        .cast(String.class)
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String String) throws Exception {
                Log.i(TAG, "accept--->" + String);
            }
        });
复制代码

测试会出现如下异常：

java.lang.ClassCastException: Cannot cast java.lang.Integer to java.lang.String
复制代码

从结果可知，发现不同类型之间转化会出现类型转化异常,cast 操作符并不能进行不同类型之间的转化，但是可以使用 cast 操作来校验发送的事件数据类型是不是指定的类型。

scan操作符

scan 操作符会依次扫描每两个元素，第一个元素没有上一个元素，则第一个元素的上一个元素会被忽略，当扫描第二个元素时，会获取到第一个元素，之后 apply 方法的返回值会作为上一个元素的值参与计算，最终返回转化后的结果，scan 官方示意图如下：

看一下下面的事件发送过程，第一次扫描时，第一个元素是 1，这里相当于 last，第二个元素是 2 ，这里相当于 item，此时 apply 方法返回的结果是 2，这个 2 会作为 last 的值参与下一次扫描计算，则下一次返回的值肯定是 2 * 3，也就是 6，测试代码如下：

Observable.just(1, 2, 3, 4, 5)
        .scan(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer last, Integer item) throws Exception {
                Log.i(TAG, "accept--last->" + last);
                Log.i(TAG, "accept--item->" + item);
                return last * item;
            }
        })
        .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "accept--->" + integer);
            }
        });
复制代码

上述代码的执行结果如下：

accept--->1
accept--last->1
accept--item->2
accept--->2
accept--last->2
accept--item->3
accept--->6
accept--last->6
accept--item->4
accept--->24
accept--last->24
accept--item->5
accept--->120
复制代码

To操作符

toList()

toList 操作符会将发送的一系列数据转换成 List，然后一次性发送出去，toList 的官方示意图如下：

测试代码如下：

Observable.just(1, 2, 3, 4)
        .toList()
        .subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> integers) throws Exception {
                Log.i(TAG, "accept--->" + integers);
            }
        });
复制代码

上述代码的执行结果如下：

accept--->[1, 2, 3, 4]
复制代码

toMap(keySelector)

toMap操作符会将要发送的事件按照指定的规则转化为 Map 形式，然后一次性发送出去，toMap 操作符官方示意图如下：

测试代码如下：

Observable.just(1, 2, 3, 4)
        .toMap(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "key"+integer;
            }
        })
        .subscribe(new Consumer<Map<String, Integer>>() {
            @Override
            public void accept(Map<String, Integer> map) throws Exception {
                Log.i(TAG, "accept--->" + map);
            }
        });
复制代码

上述代码的执行结果如下：

accept--->{key2=2, key4=4, key1=1, key3=3}
复制代码

个人微信公众号：jzman-blog 可以一起交流学习!