RxJava高级进阶--lift操作符

之前几篇文章是在为这篇文章作铺垫。关于RxJava的核心思想其实可以说就在于 lift() 。

这篇文章只分析 lift 的精髓。

其实RxJava的设计者认为开发者不应该亲自去设计 rx api，因为在不理解 lift 的情况下非常容易导致难以分析的错误。 这也是为什么使用RxJava的人会发现这玩意提供的默认api竟然有那么多，而且有些还基本长的差不多。 比如just/from，这俩基本是一回事。 这些封装就是为了最大幅度的覆盖开发中的场景，避免开发者自定义api。

学习lift的过程有点像学习Android的Framework。不深入研究它一点问题也没有，熟悉基本原理照样能写出很好的app。但是当你要做一些骚操作的话就发现fw是一个绕不过去的桩。而在熟悉了framework之后，你会发现自己的技术突然有了突飞猛进的变化。可能这就是有些开发者所说的内功修炼吧。

闲言少叙，书归正传。

回顾flatmap

之前我们说过flatmap是研究 lift 的切入点，也因此花了大量时间去说它。现在到了发挥它的光和热的时候了。 先看下面这段简单的代码，它将一个数组中的字符串派发给 observer。

String[] provinces = new String[]{"Guangdong", "Chongqing", "Shandong"};
Observer<String> observer = new Observer<String>() {
    ....
    @Override
    public void onNext(String o) {
        //GET provinces
    }
};
Observable.from(provinces)
        .subscribe(observer);

这里插句话，我们这里不用之前的weather的demo的原因是为了减少多余的代码对理解 lift 的影响。

现在抽象一个层次理解这段代码。provinces是异步源，observer是原始的接收者，它会接收到各个省的字符串。 现在我们要修改一下逻辑，让observer接收到的是各个省下面的所有城市，该怎么办？

这里就用到 flatmap，我们之前说过flatmap可以理解为一对多的变换，修改后的代码变成下面这样。

private Observable<String> getCitiesFromProvince(String province) {
    ...
}
Observable.from(provinces)
        .flatMap(new Func1<String, Observable<String>>() {
            @Override
            public Observable<String> call(String s) {
                return getCitiesFromProvince(s);
            }
        })
        .subscribe(observer);

这里定义了一个方法 getCitiesFromProvince， 它是个异步调用，返回某个省下面的所有城市， 代码稍微不很严谨，实际开发中它应该是个异步方法，需要在链式调用中加上线程切换。这里也为了方便理解省去这些步骤。

现在思考一个问题，observer所订阅的还是原来的异步源吗？

lift 在 flatmap 中的作用

这是个很有意思的问题，原始异步源没变，最终的observer也没变，但是他们的订阅关系改变了吗？

当然改变了。 展开来说，observe所订阅的不再是原始的异步源了，在原始异步源和observer中间插入了一个lift操作， lift生成一个新的observer和observable， 为了方便理解这里称为 代理异步源 和 代理接受者， 原始observer所订阅的是代理异步源， 原始异步源所派发的目标则变成了代理接受者。

上面这段话是全文的重点，理解了它也就理解了 lift 干了什么事。我发现很多被 lift 劝退的RxJava学习者都是因为现有的学习资料对 lift 的解释不够清晰直观。如果一上来就先从代码去理解它，可能会事倍功半。

下面的动图能帮助理解lift。

lift 原理

把上面 flatmap 的源码精简之后是下面这样一个调用逻辑，

public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
    ...
    return merge(map(func));
}

public static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
    ...
    return source.lift(OperatorMerge.<T>instance(false));
}

可以看到最后会调用 lift 做一个变换，注意 lift 的主体source不是原来的异步源(observable)， 而是 map(func()) 所生成的 observable，也就是代理异步源。

lift的源码有点复杂，我们把它精简一下

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);//<-- onSubscribe.call()调用的是原始异步源的call
        }
    });
}

这里分两部分解析， lift的返回__ __lift中 call 里面的 newSubscriber

lift的返回是一个 Observable 对象。结合上文所说的，这就是生成的代理异步源，我们原始的 observer 所订阅的对象会变成代理异步源。

newSubscriber是什么呢？ 其实 newSubscriber 就是上文说的代理接受者。 注意注释的那行代码，

onSubscribe.call(newSubscriber);

不要被名字迷惑，onSubscribe.call 是个接口调用，onSubscribe就是原始异步源。 也即是说，这里调用了原始异步源的 call，把原始异步源和newSubscriber做一个绑定， 在这之后，原始异步源会把结果发给代理接受者，也就是 newSubscriber。

为什么不建议用 lift

虽然 lift 也是开放api的其中一个，但是设计者不建议开发者对它做扩展。

有的人就要喷我了，看了这么长的一篇东西结果说不建议用？逗我么？ 要明白这篇东西的目的是理解RxJava的核心变换，而不是学习怎么用 lift()扩展自定义操作符。

从我的理解来说，不建议用lift的其中一个原因是它会导致流式代码的阅读性下降。 怎么理解这句话呢，比如看下面的代码

observable.map(...).filter(...).take(5).lift(new OperatorA()).subscribe(new Subscriber())

看起来这是个常见的RxJava流式调用， 数据从左到右流动，但！ 这个理解是错的。

不能明白？我教你。 还记得 lift 会产生一个新的 Observable吗？看看 lift()的返回值。

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
}

首先把 lift 的左边和右边看成两个部分。 lift() 的右边订阅的是 lift产生的 代理异步源，也就是这串东西，

observable.map(...).filter(...).take(5).lift(new OperatorA())

把它看成一个整体，上面的代码简化成一个代理异步源

proxyObservable.subscribe(new Subscriber())

然后还记得代理异步源负责接收原始异步源派发的数据这句话吗， 也就是说左边这个 proxyObservable(代理异步源),它的数据是从lift里面的 new OperatorA() 中来的。

绕晕了？ 所以非常不建议使用 lift 做骚操作。

听说过下流吗

这里的下流不是那种下流啦… RxJava中的流有上流和下流的概念，当你对RxJava有足够的了解就会涉及到这个东西。 比如这段代码，

observable.map(...).filter(...).take(5).subscribe(new Subscriber())

这就是一个标准的从左到右的流。

但加入lift之后就不一样了，

observable.map(...).filter(...).take(5).lift(new OperatorA()).subscribe(new Subscriber())

把lift()左右分开，lift()的右边是一个常规的从左到右的流，也叫下流， lift()的左边则是一个从右到左的流，也叫上流。

在RxJava中有两个专门的名词用来描述这种关系，

• UpStream
• DownStream 所以以后你看到UpStream和DownStream就明白是怎么回事了吧。