响应式编程3数据处理(RxJava2.x示例)

除了观察者模式原理，响应式编程的另一个重要内容就是数据的处理。RxJava提供了类似java8中stream流式的数据处理方式，可以对被观察对象进行处理。

RxJava数据处理实例

    @Test
    public void test6() {
        Observable.just(-1, -2, -2, -3, -4, 5, 5, 6, 7, 8) // 事件发生
//                .observeOn(Schedulers.newThread())  // 指定事件发生的线程
//                .subscribeOn(Schedulers.io())  // 指定事件回调的线程
                .distinct() // 去掉重复的：[-1, -2, -3, -4, 5, 6, 7, 8]
                .filter(i -> i % 2 != 0) // 去掉偶数：[-1, -3, 5, 7]
                .map(Math::abs) // 取绝对值：[1, 3, 5, 7]
                .buffer(2, 2)// 分成多个buffer组：
                                        // count为每个buffer中元素的个数，
                                        // skip为下一个分组相对偏移的元素个数
                                        // 结果为：[1, 3], [5, 7]
                                        // 可以调整为 buffer(2, 1) 试试效果
                .subscribe(
                        System.out::println, // 事件成功的处理 （onNext）
                        e -> System.out.println(e.getMessage()) // 事件失败的处理 （onError）
                );

    }

数据处理详解

• distinct 去掉重复的
• filter 过滤器，接收一个Predicate类型的实例，用来判断是否过滤
• map 映射器，接受一个Function，作为映射的映射关系（转换关系）
• buffer 分组，接收两个整型参数count和skip，count表示每组的元素个数、skip表示每次分组起始元素的偏移量。

最终subscribe方法出现的数据就是经过以上几步处理过的数据。