频道栏目
首页 > 资讯 > Java > 正文

快速理解RxJava源码的设计理念

16-05-05        来源:[db:作者]  
收藏   我要投稿

前言

我在看过几篇关于RxJava源码分析的博客后,不知是我的水平有限还是源码过于博大精深,导致花了很长的时间才搞清楚其运行原理。我个人觉得应该有更好的办法来快速剖析理解,于是决定写下本文。
本文适合已经看过一些RxJava源码分析资料的同学,不过没看过也没关系。

一、初探RxJava

我们先来看一段最基本的代码,分析这段代码在RxJava中是如何实现的。

Observable.OnSubscribe onSubscriber1 = new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext("1");
        subscriber.onCompleted();
    }
};
Subscriber subscriber1 = new Subscriber() {
    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(String s) {

    }
};

Observable.create(onSubscriber1)
        .subscribe(subscriber1);

首先我们来看一下Observable.create的代码:

public final static  Observable create(OnSubscribe f) {
    return new Observable(hook.onCreate(f));
}

protected Observable(OnSubscribe f) {
    this.onSubscribe = f;
}

直接就是调用了Observable的构造函数来创建一个新的Observable对象,这个对象我们暂时标记为observable1,以便后面追溯。
同时,会将我们传入的OnSubscribe对象onSubscribe1保存在observable1的onSubscribe属性中,这个属性在后面的上下文中很重要,大家留心一下。

接下来我们来看看subscribe方法。

public final Subscription subscribe(Subscriber subscriber) {
    return Observable.subscribe(subscriber, this);
}

private static  Subscription subscribe(Subscriber subscriber, Observable observable) {
    ...
    subscriber.onStart();
    hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
    return hook.onSubscribeReturn(subscriber);
}

可以看到,subscribe之后,就直接调用了observable1.onSubscribe.call方法,也就是我们代码中的onSubscribe1对象的call方法
,传入的参数就是我们代码中定义的subscriber1对象。call方法中所做的事情就是调用传入的subscriber1对象的onNext和onComplete方法。
这样就实现了观察者和被观察者之间的通讯。
【摘录结束】
好了,总结一下:

处理数据是订阅类Subscriber的职责

入门结束,下面看看带操作符的情况。

二、带操作符的RxJava应用

先展示Demo:

Observable.OnSubscribe onSubscriber1 = new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext("1");
        subscriber.onCompleted();
    }
};
Subscriber subscriber1 = new Subscriber() {
    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Integer i) {

    }
};
Func1 transformer1 = new Func1() {
    @Override
    public Integer call(String s) {
        return Integer.parseInt(s);
    }
};

Observable.create(onSubscriber1)
        .map(transformer1)
        .subscribe(subscriber1);

和刚才不同的是我们在create之后调用了map方法,然后才调用subscribe方法。
【摘录结束】
再贴上RxJava的执行流程图:
lift流程图

我先根据这张图给出RxJava能这样运行的3条准则:

从create函数能向下执行到subscribe,靠的是返回Observable类的对象; 从subscribe函数能再向上执行到onSubscriber1,靠的是下面的Observable对象能够调用上面Observable对象的onSubscribe属性; 从上面能再次往下依次对数据进行处理,靠的是Subscriber责任链。

为什么我不先分析源码而先给准则呢?我发现如果在不知道准则(或者称作设计思路)情况下看源码就会“见树不见林”,而知道了准则就把握了框架脉络,剩下的只是到源码中找答案而已。既然本文标榜了“快速理解RxJava源码的设计理念”,那当然不能徒有虚名。下面我们到源码中找答案。

首先我们来证明准则1:

map方法的代码如下:

public final  Observable map(Func1 func) {
    return lift(new OperatorMap(func));
}

map返回的是Observable,证明完毕。快吧,就这么快。在此将map返回的Observable对象标记为observable2。

再证明准则2:

对照demo中的代码,我们调用map之后,就调用了subscribe方法,也就是调用了这里的observable2的subscribe方法。
根据上面的介绍,调用subscribe之后,就会调用observable2.onSubscribe.call方法:

public final  Observable lift(final Operator operator) {
    return new Observable(new OnSubscribe() {
        @Override
        public void call(Subscriber o) {
            Subscriber st = hook.onLift(operator).call(o);
            st.onStart();
            onSubscribe.call(st);
        }
    });
}

其中,call函数的第一句与准则3的证明有关,暂时不看。第二句也不看(不好意思,我没看懂这句干什么的,不过不影响继续分析),我们看第三句:

onSubscribe.call(st);

这句很关键,准则2的证明靠他了。我们要搞明白onSubscribe是谁的属性。我将demo换个方式写出来:

...
Observable.create(onSubscriber1);
observable1.map(transformer1);
observable2.subscribe(subscriber1);

map函数是observable1的,那么lift函数也是observable1的,按Java语法规定内部类可以调用外部类的属性,所以onSubscribe是observable1的。这样就实现了observable2跨入到observable1的目的。证毕。

最后证明准则3:

说实话,写到这里我感觉还是有部分同学没明白怎么回事。不过没关系,我将会继续围绕讲述。
我们再来看看observable2.onSubscribe.call方法:

public final  Observable lift(final Operator operator) {
    return new Observable(new OnSubscribe() {
        @Override
        public void call(Subscriber o) {
            Subscriber st = hook.onLift(operator).call(o);
            st.onStart();
            onSubscribe.call(st);
        }
    });
}

前面说了,call函数的第一句与准则3的证明有关。我们先不看它做了什么,返回值给了st,st是个Subscriber对象。我们在第一节总结说了,处理数据是订阅类Subscriber的职责,因此st有处理数据的职责。再看第三句,将st给了observable1。问题来了,准则3想要运行,就必须有Subscriber责任链。“链”在哪里?不用猜了,我直接告诉你答案:往下“链”的代码在:

Subscriber st = hook.onLift(operator).call(o);

我们看看OperatorMap的call方法:

@Override
public Subscriber call(final Subscriber o) {
    return new Subscriber(o) {
        @Override
        public void onNext(T t) {
            o.onNext(transformer.call(t));
        }
    };
}

入口参数o在Demo里就是subscriber1对象。我们看它的这句:

o.onNext(transformer.call(t));

按照语法规定,是先执行括号里的。我把它写得再直白一点:

临时数据=map(处理前的数据);//本步骤处理
subscriber1.onNext(临时数据);//提交给下一步处理

往上“链”的代码在:

onSubscribe.call(st);

总结一下:
在observable2这种操作符返回的被观察者中,OnSubscribe要做两件事:

在本级中生成订阅类Subscriber,在Subscriber中将本级的操作(map)与下一级操作连接起来(向下链接) 将本级中生成订阅类Subscriber交给上一级Observable,方便上一级向下链接

因此,在此的订阅类Subscriber的职责实际上有两个:

处理数据 连接下一个步骤的Subscriber
再上一张图帮助消化:
这里写图片描述

后记:RxJava为什么这么搞

相信你对lift流程图挺有印象:

lift流程图

我看到后第一个感觉就是:有必要搞得这么复杂吗?幸好还记得设计模式中的装饰者模式和责任链模式,还能够大体上理解。不过老问题依然存在:它这么搞有必要吗?学习一个陌生的事物最好的办法就是拿一个熟悉的事物进行对比,回想一下软件在运行前会进行什么工作?答案是配置。在代码

Observable.create(onSubscriber1)
          .map(transformer1)
          .subscribe(subscriber1);

create也好,map也罢,都是配置。再问个问题,软件配置好了怎么运行?要么是点击软件,要么是命令启动。subscribe就是启动。好了,我们可以进行抽象了。无论是RxJava还是软件,其核心就是“配置+启动”。说了RxJava与软件的相同之处,它俩有什么不同吗?你看软件是先配置再启动对吧,没见过还要像”lift流程图”一样上串下跳吧。这就是RxJava的设计理念不同了。我当时在想,如果我来设计,我会先create一个Observable对象,然后将map这类的操作符放到Observable对象里的一个”操作队列”里面,这样配置就完成了。等需要启动时,调用subscribe启动”操作队列”处理数据,你看代码执行流程不是很简单吗。这种实现办法是通过数据结构达成的,优点就是源码执行流程清楚简单,缺点就是这个对象职责过多。而RxJava是通过责任链模式达成的,优点是职责分散,缺点就是源码执行起来上串下跳。不过RxJava在设计时已经为用户考虑到这点,实际上如果你不看源码只是使用RxJava,对你而言代码的执行就是一条线下来也很清晰。好了,就讲到这吧,希望对理解RxJava能有所帮助。

 

相关TAG标签
上一篇:基于wsdl的测试数据自动生成技术
下一篇:HTML 5 视频
相关文章
图文推荐

关于我们 | 联系我们 | 广告服务 | 投资合作 | 版权申明 | 在线帮助 | 网站地图 | 作品发布 | Vip技术培训 | 举报中心

版权所有: 红黑联盟--致力于做实用的IT技术学习网站