RxJava 1.x 理解-3

Posted 2020-10-24 H_bolin

在 RxJava 1.x 理解-1 中，我们说到了RxJava的简单用法，但是这还远远不够，因为

输入的数据 ---> 被监听者（订阅源）对这些数据进行操作，或者执行响应的处理 --> 产生新的数据，或者事件发送给监听者 --> 监听者执行自己的方法。

其中，RxJava还可以对输入的数据进行变换，产生新数据（可以是复杂的数据），而不是简单的事件触发。

先将数据的提供，进阶一下：

rxJava just 使用

    /**
     * rxJava just 使用
     * just --> 还是使用的create方法
     */
    private void rxJavaJust() {
        Log.d(TAG, "----------- just ---------");

        Observable
                .just("just 1", "just 2", "just 3")
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.d(TAG, "Item: " + s);
                    }
                });
    }

just会不断的将可变参数数据 just 1 ；just 2 ；just 3 .... 发送出去

rxJava from 使用

    /**
     * rxJava from 使用
     * from --> 还是使用的create方法
     */
    private void rxJavaFrom() {
        Log.d(TAG, "----------- from ---------");

        String[] names = {"1", "2", "3"};
        Observable.from(names)
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onNext(String s) {
                        Log.d(TAG, "Item: " + s);
                    }

                    @Override
                    public void onCompleted() {
                        Log.d(TAG, "Completed!");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "Error!");
                    }
                });

        Log.d(TAG, "----------- from2 ---------");

        // 简单来说就是数据提供了，而且是一个个的提供了，至于要如何执行，那就按观察者自己的事情了。
        Observable.from(names)
                .subscribe(new Action1<String>() {

                    @Override
                    public void call(String s) {
                        Log.d(TAG, "Item: " + s);
                    }

                });
    }

from会不断的将数组或者集合中的数据一个个发送出去

数据变换

rxjava map 变换

    /**
     * rxjava map 变换
     * 将类型转化成另一个类型
     */
    private void rxJavaMap() {
        Log.d(TAG, "----------- Map ---------");

        Observable.just("1") // 输入类型 String
                .map(new Func1<String, Integer>() {
                    @Override
                    public Integer call(String s) {
                        return Integer.parseInt(s);
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.d(TAG, "Item: " + integer + " 执行调用 获取线程id:" + Thread.currentThread().getId());
                    }
                });
    }

        Observable.just("999", "11") // 输入类型 String
                .map(new Func1<String, List<Integer>>() {
                    @Override
                    public List<Integer> call(String s) {
                        List<Integer> ints = new ArrayList<>();
                        ints.add(Integer.parseInt(s));
                        ints.add(Integer.parseInt(s));
                        ints.add(Integer.parseInt(s));
                        return ints;
                    }
                }).subscribe(new Action1<List<Integer>>() {
            @Override
            public void call(List<Integer> integers) {
                int i = 0;
                // 这里进行遍历
                for (Integer integer : integers) {
                    Log.e(TAG, "Item " + i + " :" + +integer);
                    i += 1;
                }
            }
        });

    }

可以看到，将String格式的数据转换成了Integer格式的数据，或者是String格式的数据转换成List<Integer>格式的数据。map是一对一的变换。

rxjava flatMap 变换

    /**
     * rxjava flatMap 变换
     * 事件再次分发：
     *
     * 从上面的代码可以看出， flatMap() 和 map() 有一个相同点：它也是把传入的参数转化之后返回另一个对象。
     * 但需要注意，和 map() 不同的是， flatMap() 中返回的是个 Observable 对象，但这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。
     * flatMap() 的原理是这样的：
     * 1. 使用传入的事件对象创建一个 Observable 对象；
     * 2. 并不发送这个 Observable, 而是将它激活，于是它开始发送事件；
     * 3. 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
     * 这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。而这个『铺平』就是 flatMap() 所谓的 flat。
     */
    private void rxJavaFlatMap() {
        Log.d(TAG, "----------- rxJavaFlatMap ---------");

        List<Student> students = new ArrayList<>();
        List<Source> sources = new ArrayList<>();
        sources.add(new Source(1, "化学", 80));
        sources.add(new Source(2, "物理", 79));
        sources.add(new Source(3, "生物", 78));
        students.add(new Student("小明1", 1, sources));
        students.add(new Student("小明2", 1, sources));
        Observable.from(students)
                .flatMap(new Func1<Student, Observable<Source>>() {
                    @Override
                    public Observable<Source> call(Student student) {
                        Log.d(TAG, "Item: " + student.name);
                        return Observable.from(student.mSources);
                    }
                })
                .subscribe(new Action1<Source>() {
                    @Override
                    public void call(Source source) {
                        Log.d(TAG, "Item: " + source + " 执行调用 获取线程id:" + Thread.currentThread().getId());
                    }
                });
    }

 // ------------------------ 测试使用的类 -----------------------------

    class Student {
        String name;//学生名字
        int id;
        List<Source> mSources;//每个学生的所有课程

        public Student(String name, int id, List<Source> sources) {
            this.name = name;
            this.id = id;
            mSources = sources;
        }

        @Override
        public String toString() {
            return "Student{" +
                    "name=\'" + name + \'\\\'\' +
                    ", id=" + id +
                    ", mSources=" + mSources +
                    \'}\';
        }
    }

    class Source {
        int sourceId;//id
        String name;//课程名
        int score;//成绩

        public Source(int sourceId, String name, int score) {
            this.sourceId = sourceId;
            this.name = name;
            this.score = score;
        }

        @Override
        public String toString() {
            return "Source{" +
                    "sourceId=" + sourceId +
                    ", name=\'" + name + \'\\\'\' +
                    ", score=" + score +
                    \'}\';
        }
    }

输出结果：

02-09 15:45:50.580 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: 小明1
02-09 15:45:50.581 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: Source{sourceId=1, name=\'化学\', score=80} 执行调用 获取线程id:2
02-09 15:45:50.581 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: Source{sourceId=2, name=\'物理\', score=79} 执行调用 获取线程id:2
02-09 15:45:50.582 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: Source{sourceId=3, name=\'生物\', score=78} 执行调用 获取线程id:2
02-09 15:45:50.582 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: 小明2
02-09 15:45:50.582 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: Source{sourceId=1, name=\'化学\', score=80} 执行调用 获取线程id:2
02-09 15:45:50.582 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: Source{sourceId=2, name=\'物理\', score=79} 执行调用 获取线程id:2
02-09 15:45:50.583 3223-3223/pers.bolin.rxjavademo D/MainActivity: Item: Source{sourceId=3, name=\'生物\', score=78} 执行调用 获取线程id:2

 其实上述代码，也可以用map实现：

        Observable.from(students).map(new Func1<Student, List<Source>>() {
            @Override
            public List<Source> call(Student student) {
                Log.d(TAG, "Item: " + student.name);
                return student.mSources;
            }
        }).subscribe(new Action1<List<Source>>() {
            @Override
            public void call(List<Source> sources) {
                // 主要的差距是这里，我们自己进行了一次循环遍历。而flatMap则不用，直接获取到的就是Source对象，而不是List<Source>对象
                for (Source source:sources){
                    Log.d(TAG, "Item: " + source + " 执行调用 获取线程id:" + Thread.currentThread().getId());
                }
            }
        });

从上面的代码可以看出， 
flatMap() 和 map() 有一个相同点：它也是把传入的参数转化之后返回另一个对象。
但需要注意，和 map() 不同的是， flatMap() 中返回的是个 Observable 对象，但这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。
flatMap() 的原理是这样的：
1. 使用传入的事件对象创建一个 Observable 对象；
2. 并不发送这个 Observable, 而是将它激活，于是它开始发送事件；
3. 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。而这个『铺平』就是 flatMap() 所谓的 flat。

总结：RxJava的作用就是
对于数据：对输入的数据进行变换，转换成想要的数据，并且可以指定执行的线程，得到最终的结果。
对于事件：触发事件，观察者感知，观察者执行操作，并且可以指定执行的线程。

参考资料：

一起来造一个RxJava，揭秘RxJava的实现原理

给 Android 开发者的 RxJava 详解