Android 开发者的 RxJava 详解(四)

4. 变换

终于要到牛逼的地方了，不管你激动不激动，反正我是激动了。

RxJava 提供了对事件序列进行变换的支持，这是它的核心功能之一，也是大多数人说『RxJava 真是太好用了』的最大原因。所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。概念说着总是模糊难懂的，来看 API。

4.1 API
首先看一个 map()的例子：

Observable.just("images/logo.png") // 输入类型 String
    .map(new Func1<String, Bitmap>() {
        @Override
        public Bitmap call(String filePath) { // 参数类型 String
            return getBitmapFromPath(filePath); // 返回类型 Bitmap
        }
    })
    .subscribe(new Action1<Bitmap>() {
        @Override
        public void call(Bitmap bitmap) { // 参数类型 Bitmap
            showBitmap(bitmap);
        }
    });

这里出现了一个叫做Func1的类。它和Action1非常相似，也是 RxJava 的一个接口，用于包装含有一个参数的方法。Func1和Action 的区别在于，Func1包装的是有返回值的方法。另外，和ActionX 一样，FuncX也有多个，用于不同参数个数的方法。FuncX和 ActionX 的区别在FuncX包装的是有返回值的方法。

可以看到，map()方法将参数中的 String 对象转换成一个 Bitmap 对象后返回，而在经过map()方法后，事件的参数类型也由String转为了Bitmap。这种直接变换对象并返回的，是最常见的也最容易理解的变换。不过 RxJava 的变换远不止这样，它不仅可以针对事件对象，还可以针对整个事件队列，这使得 RxJava 变得非常灵活。我列举几个常用的变换：

• map(): 事件对象的直接变换，具体功能上面已经介绍过。它是 RxJava 最常用的变换。map() 的示意图：

  

• flatMap(): 这是一个很有用但非常难理解的变换，因此我决定花多些篇幅来介绍它。 首先假设这么一种需求：假设有一个数据结构『学生』，现在需要打印出一组学生的名字。实现方式很简单：

Student[] students = ...;
Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String name) {
        Log.d(tag, name);
    }
    ...
};
Observable.from(students)
    .map(new Func1<Student, String>() {
        @Override
        public String call(Student student) {
            return student.getName();
        }
    })
    .subscribe(subscriber);

很简单。那么再假设：如果要打印出每个学生所需要修的所有课程的名称呢？（需求的区别在于，每个学生只有一个名字，但却有多个课程。）首先可以这样实现：

Student[] students = ...;
Subscriber<Student> subscriber = new Subscriber<Student>() {
    @Override
    public void onNext(Student student) {
        List<Course> courses = student.getCourses();
        for (int i = 0; i < courses.size(); i++) {
            Course course = courses.get(i);
            Log.d(tag, course.getName());
        }
    }
    ...
};
Observable.from(students)
    .subscribe(subscriber);

依然很简单。那么如果我不想在Subscriber中使用 for 循环，而是希望 Subscriber中直接传入单个的Course对象呢（这对于代码复用很重要）？用map()显然是不行的，因为 map() 是一对一的转化，而我现在的要求是一对多的转化。那怎么才能把一个 Student 转化成多个Course呢？
这个时候，就需要用 flatMap() 了：

Student[] students = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() {
    @Override
    public void onNext(Course course) {
        Log.d(tag, course.getName());
    }
    ...
};
Observable.from(students)
    .flatMap(new Func1<Student, Observable<Course>>() {
        @Override
        public Observable<Course> call(Student student) {
            return Observable.from(student.getCourses());
        }
    })
    .subscribe(subscriber);

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

传统的嵌套请求需要使用嵌套的 Callback 来实现。而通过 flatMap() ，可以把嵌套的请求写在一条链中，从而保持程序逻辑的清晰。

• throttleFirst(): 在每次事件触发后的一定时间间隔内丢弃新的事件。常用作去抖动过滤，例如按钮的点击监听器：RxView.clickEvents(button)//RxBinding代码，后面的文章有解释 .throttleFirst(500, TimeUnit.MILLISECONDS)// 设置防抖间隔为 500ms .subscribe(subscriber); 妈妈再也不怕我的用户手抖点开两个重复的界面啦。
  此外， RxJava 还提供很多便捷的方法来实现事件序列的变换，这里就不一一举例了。

4.2 变换的原理：lift()

这些变换虽然功能各有不同，但实质上都是针对事件序列的处理和再发送。而在RxJava 的内部，它们是基于同一个基础的变换方法： lift(Operator)。首先看一下lift()的内部实现（仅核心代码）：

// 注意：这不是 lift() 的源码，而是将源码中与性能、兼容性、扩展性有关的代码剔除后的核心代码。
// 如果需要看源码，可以去 RxJava 的 GitHub 仓库下载。
public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
    return Observable.create(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber subscriber) {
            Subscriber newSubscriber = operator.call(subscriber);
            newSubscriber.onStart();
            onSubscribe.call(newSubscriber);
        }
    });
}

这段代码很有意思：它生成了一个新的 Observable 并返回，而且创建新 Observable 所用的参数 OnSubscribe 的回调方法 call() 中的实现竟然看起来和前面讲过的 Observable.subscribe() 一样！然而它们并不一样哟~不一样的地方关键就在于第二行 onSubscribe.call(subscriber) 中的 onSubscribe 所指代的对象不同（高能预警：接下来的几句话可能会导致身体的严重不适）——

• subscribe() 中这句话的onSubscribe指的是Observable中的 onSubscribe对象，这个没有问题，但是lift()之后的情况就复杂了点。
• 当含有lift()时：
  1.lift()创建了一个Observable后，加上之前的原始Observable，已经有两个Observable了；
  2.而同样地，新Observable里的新OnSubscribe加上之前的原始 Observable中的原始OnSubscribe，也就有了两个OnSubscribe；
  3.当用户调用经过lift()后的Observable的subscribe()的时候，使用的是lift()所返回的新的Observable ，于是它所触发的 onSubscribe.call(subscriber)，也是用的新Observable中的新 OnSubscribe，即在lift()中生成的那个OnSubscribe；
  4.而这个新OnSubscribe的call()方法中的onSubscribe，就是指的原始Observable中的原始OnSubscribe，在这个call()方法里，新 OnSubscribe利用operator.call(subscriber) 生成了一个新的 Subscriber（Operator 就是在这里，通过自己的call()方法将新 Subscriber和原始 Subscriber进行关联，并插入自己的『变换』代码以实现变换），然后利用这个新Subscriber向原始Observable进行订阅。
  这样就实现了lift()过程，有点像一种代理机制，通过事件拦截和处理实现事件序列的变换。

精简掉细节的话，也可以这么说：在Observable执行了lift(Operator)方法之后，会返回一个新的Observable，这个新的Observable会像一个代理一样，负责接收原始的Observable发出的事件，并在处理后发送给Subscriber。
如果你更喜欢具象思维，可以看图：

或者可以看动图：

两次和多次的lift()同理，如下图：

举一个具体的Operator的实现。下面这是一个将事件中的Integer对象转换成String的例子，仅供参考：

observable.lift(new Observable.Operator<String, Integer>() {
    @Override
    public Subscriber<? super Integer> call(final Subscriber<? super String> subscriber) {
        // 将事件序列中的 Integer 对象转换为 String 对象
        return new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                subscriber.onNext("" + integer);
            }

            @Override
            public void onCompleted() {
                subscriber.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                subscriber.onError(e);
            }
        };
    }
});

讲述lift()的原理只是为了让你更好地了解 RxJava ，从而可以更好地使用它。然而不管你是否理解了lift()的原理，RxJava 都不建议开发者自定义 Operator来直接使用lift()，而是建议尽量使用已有的lift()包装方法（如 map() flatMap() 等）进行组合来实现需求，因为直接使用lift()非常容易发生一些难以发现的错误。

4.3 compose: 对 Observable 整体的变换

除了lift()之外，Observable还有一个变换方法叫做 compose(Transformer)。它和lift()的区别在于，lift()是针对事件项和事件序列的，而compose()是针对Observable自身进行变换。举个例子，假设在程序中有多个Observable，并且他们都需要应用一组相同的lift()变换。你可以这么写：

observable1
    .lift1()
    .lift2()
    .lift3()
    .lift4()
    .subscribe(subscriber1);
observable2
    .lift1()
    .lift2()
    .lift3()
    .lift4()
    .subscribe(subscriber2);
observable3
    .lift1()
    .lift2()
    .lift3()
    .lift4()
    .subscribe(subscriber3);
observable4
    .lift1()
    .lift2()
    .lift3()
    .lift4()
    .subscribe(subscriber1);

你觉得这样太不软件工程了，于是你改成了这样：

private Observable liftAll(Observable observable) {
    return observable
        .lift1()
        .lift2()
        .lift3()
        .lift4();
}
...
liftAll(observable1).subscribe(subscriber1);
liftAll(observable2).subscribe(subscriber2);
liftAll(observable3).subscribe(subscriber3);
liftAll(observable4).subscribe(subscriber4);

可读性、可维护性都提高了。可是Observable被一个方法包起来，这种方式对于Observale的灵活性似乎还是增添了那么点限制。怎么办？这个时候，就应该用compose()来解决了：

public class LiftAllTransformer implements Observable.Transformer<Integer, String> {
    @Override
    public Observable<String> call(Observable<Integer> observable) {
        return observable
            .lift1()
            .lift2()
            .lift3()
            .lift4();
    }
}
...
Transformer liftAll = new LiftAllTransformer();
observable1.compose(liftAll).subscribe(subscriber1);
observable2.compose(liftAll).subscribe(subscriber2);
observable3.compose(liftAll).subscribe(subscriber3);
observable4.compose(liftAll).subscribe(subscriber4);

像上面这样，使用compose()方法，Observable可以利用传入的 Transformer对象的call方法直接对自身进行处理，也就不必被包在方法的里面了。

compose()的原理比较简单，不附图喽。
【附录】

资料图

需要资料的朋友可以加入Android架构交流QQ群聊：513088520

点击链接加入群聊【Android移动架构总群】：加入群聊

获取免费学习视频，学习大纲另外还有像高级UI、性能优化、架构师课程、NDK、混合式开发（ReactNative+Weex）等Android高阶开发资料免费分享。