Compose操作符

RxJava中的compose方法是一种组合操作符，它可以将一系列的RxJava操作符组合在一起，形成一个新的Observable。compose方法的定义如下：

public final <R> Observable<R> compose(Transformer<? super T, ? extends R> composer) {
    return RxJavaPlugins.onAssembly(new ObservableCompose<T, R>(this, composer));
}

其中，Transformer<? super T, ? extends R>是一个变换器，用于将原始的Observable转换成一个新的Observable，类型参数T表示原始Observable发射的数据类型，类型参数R表示新的Observable发射的数据类型。

compose方法的作用是将一系列的操作符打包成一个变换器，然后应用到原始的Observable中。这个变换器可以包含任意数量的操作符，而且操作符的顺序和组合方式可以随意调整。

下面是一个使用compose方法的示例：

Observable.range(1, 5)
        .compose(applySchedulers())
        .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept: " + integer);
            }
        });

private <T> ObservableTransformer<T, T> applySchedulers() {
    return new ObservableTransformer<T, T>() {
        @Override
        public ObservableSource<T> apply(Observable<T> upstream) {
            return upstream.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread());
        }
    };
}

这个示例中，applySchedulers方法返回一个ObservableTransformer对象，它将原始的Observable转换成一个新的Observable，这个新的Observable会在io线程上执行订阅操作，然后切换到主线程上执行观察者的回调方法。这个变换器被应用到range操作符上，形成一个新的Observable，然后订阅这个新的Observable，观察者会收到一个排好序的数字序列。

使用compose方法的好处是可以将一系列的操作符打包成一个变换器，这样可以减少代码量，提高代码可读性。另外，这个变换器可以在多个Observable中重复使用，避免重复编写相同的代码。

flatMap操作符

RxJava中的flatMap方法是一种转换操作符，它可以将一系列的RxJava操作符组合在一起，形成一个新的Observable。flatMap方法的定义如下：

public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
    return flatMap(mapper, false);
}

其中，Function<? super T, ? extends ObservableSource<? extends R>>T是接受到的数据类型， ObservableSource<? extends R>是代表返回一个新的流且数据类型是R

flatMap是一个比教难理解的一个转换，在这里先假设一个需求，需要打印多个Student所学的课程。这跟之前获取Student的name又不同了，这里先确定一下关系，一个Student类中只有一个name，而一个Student却有多门课程（Course），Student我们可以理解成这样：

    /**
     * 学生类
     */
    class Student {
        private String name;//姓名
        private List<Course> coursesList;//所修的课程
        ...
    }
    /**
     * 课程类
     */
    class  Course {
        private String name;//课程名
        private String id;
        ...
    }

如果使用map来实现打印所有学生所修个课程名，实现的代码是这样的：

        List<Student> students = new ArrayList<Student>();
        students.add...
        ...

        Action1<List<Course>> action1 = new Action1<List<Course>>() {
            @Override
            public void call(List<Course> courses) {
                //遍历courses，输出cuouses的name
                 for (int i = 0; i < courses.size(); i++){
                    Log.i(TAG, courses.get(i).getName());
                }
            }
        };
        Observable.from(students)
                .map(new Func1<Student, List<Course>>() {
                    @Override
                    public List<Course> call(Student student) {
                        //返回coursesList
                        return student.getCoursesList();
                    }
                })
                .subscribe(action1);

可以看到，在Action1中出现了for来循环打印课程名，使用RxJava就是为了剔除这样的嵌套结构，使得整体的逻辑性更强。这时候就可以使用flatMap了，使用flatMap实现的代码是这样的：

        List<Student> students = new ArrayList<Student>();
        students.add...
        ...

        Observable.from(students)
                .flatMap(new Func1<Student, Observable<Course>>() {
                    @Override
                    public Observable<Course> call(Student student) {
                        return Observable.from(student.getCoursesList());
                    }
                })
                .subscribe(new Action1<Course>() {
                    @Override
                    public void call(Course course) {
                        Log.i(TAG, course.getName());
                    }
                });

这样就实现了跟上面代码一样的效果，看起来有点懵？确实，flatMap理解起来有点绕，刚接触flatMap的时候我也是懵逼一个。下面我将flatMap的示意图，希望能帮助理解：

flatMap示意图

由上图可以看出Student1、Student2经过flatMap后，按顺序依次经历了Observable1、Observable2，分别转化为Course。最后按顺序得到Course1、Course2、Course3、Course4、Course5、Course6，其中1-3由Student1得到，4-6由Student2得到。结合代码和示意图，是不是对flatMap有了一定的理解。

注意：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。也就说，传入的顺序可能跟出来的顺序不一样。如果要保证顺的的话，可以使用concatMap。

异同

compose操作符和flatMap操作符都可以生成新的流，但是compose是传入的是流本身，flatMap传入的是流中的数据。

Observable.just(1, 2, 3)
.compose(upstream ->upstream.subscribeOn(Schedulers.io()))
.subscribe(System.out::println);