RxJava学习(三)

变换

所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。

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() 示意图：

扩展：由于可以在嵌套的 Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。示例代码（Retrofit + RxJava）：

networkClient.token() // 返回 Observable<String>，在订阅时请求 token，并在响应后发送 token

    .flatMap(new Func1<String, Observable<Messages>>() {

        @Override

        public Observable<Messages> call(String token) {

            // 返回 Observable<Messages>，在订阅时请求消息列表，并在响应后发送请求到的消息列表

            return networkClient.messages();

        }

    })

    .subscribe(new Action1<Messages>() {

        @Override

        public void call(Messages messages) {

            // 处理显示消息列表

            showMessages(messages);

        }

    });

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

• throttleFirst(): 在每次事件触发后的一定时间间隔内丢弃新的事件。常用作去抖动过滤，例如按钮的点击监听器：

RxView.clickEvents(button) // RxBinding 代码，后面的文章有解释

    .throttleFirst(500, TimeUnit.MILLISECONDS) // 设置防抖间隔为 500ms

    .subscribe(subscriber);

妈妈再也不怕我的用户手抖点开两个重复的界面啦。

此外， RxJava 还提供很多便捷的方法来实现事件序列的变换，这里就不一一举例了。

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);

            }

        };

    }

});

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() 的原理比较简单。