RxJava 链式调用流程源码记录分析（以map()为例）

瞅了源码，记录下RxJava的链式调用原理。下文主要以 map()函数的调用流程分析
（若不熟悉基本使用请看此文给 Android 开发者的 RxJava 详解）

实例

public class RxJavaDemoActivity extends Activity{    @Override    protected void onCreate( Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentView(R.layout.activity_rxjavamain);        test();    }    private void test() {        Observable.create(new Observable.OnSubscribe() {            @Override            public void call(Subscriber<? super Integer> subscriber) {                subscriber.onNext(10);            }        }).map(new Func1() {            @Override            public Integer call(Integer integer) {                return integer *10;            }        }).map(new Func1() {            @Override            public Integer call(Integer integer) {                return integer + 1;            }        }).subscribe(new Subscriber() {            @Override            public void onCompleted() {            }            @Override            public void onError(Throwable e) {            }            @Override            public void onNext(Integer integer) {                Toast.makeText(RxJavaDemoActivity.this,"运行结果: "+integer,Toast.LENGTH_LONG).show();            }        });    }}

以上代码运行结果为：101
运行逻辑流程为：
1. subscriber.onNext(10); 相当于输入参数为 int a = 10；
2. return integer *10; 对第一步的输入扩大10倍，a = a * 10 = 100
3. return integer + 1; 对第二步结果加1， a = a + 1 = 101
4. Toast.makeText(RxJavaDemoActivity.this,"运行结果: "+integer,Toast.LENGTH_LONG).show(); 最终结果通过 Toast显示出来

步骤分析

流程图示
先给出实例代码的总体流程图吧（图示中类名采用了简写）

下面分成两部分展开：
一、第一部分：

（注：对 .map(new Func1() 中的Funct1加了后缀数字标识）
步骤分析：
1、下面语句会创建一个Observable（图中描述为Ob1），并为 Ob1 创建了Onsubscribe（图中描述为 On1）

 Observable.create(new Observable.OnSubscribe() {     @Override     public void call(Subscriber<? super Integer> subscriber) {         subscriber.onNext(10);     } })

2、执行第一个map(Func1 )，图中以 Func1_2 标识该 Func1。查看map()函数的源码可以看到这步生成了一个新的 Observable （记为 Ob2），并为Ob2 创建了新的 Onsubscribe（记为On2）。这里需要注意的是新的On2包含了两部分的内涵：On1 和此处传入的Func1（即Func1_2），即新的Onsubscribe包含了先前的Onsubscribe以及map方法传入的Func1处理。本地执行完返回的是新的Observable（Ob2）

.map(new Func1<Integer, Integer>() {    @Override    public Integer call(Integer integer) {        return integer *10;    }})

3、执行第二个map()，以Func1_3 标识传入的Func1。同上，会生成一个新的Observable（Ob3），并未Ob3新建了OnSubscribe（On3）。同样，On3包含了两部分内涵：On2 和此处传入的 Func1_3。执行完返回新的Observable（Ob3）

.map(new Func1<Integer, Integer>() {    @Override    public Integer call(Integer integer) {        return integer + 1;    }})

至此，最终得到的是 Ob3

二、第二部分

步骤分析：
1、先来看 subscribe()运行时的主要操作

static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {        ...        subscriber.onStart();        ...        try {            // allow the hook to intercept and/or decorate            hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);            return hook.onSubscribeReturn(subscriber);        }  }

hook.onSubscribeStart(observable, observable.onSubscribe) ：原样返回参数observable.onSubscribe。所以，subscribe(),实际上相当于
Ob3.On3.call(subscriber) ,subscriber 即为本步骤传入的Subscriber（记为Sb）

2、On3.call(Sb)
对于map()方法而言，On3 实际上是 OnSubscribeLift,继承自 OnSubscribe

public final class OnSubscribeLift<T, R> implements OnSubscribe<R> {    ...    final OnSubscribe parent;//上一级被包含的OnSubscribe    final Operator<? extends R, ? super T> operator;//operator包含了Func1()    ...    @Override    public void call(Subscriber<? super R> o) {        try {            //生成新的 Subsciber，该Subscriber包含本OnSubscribeLift拥有的Func1 和 传入的 Subscriber o            Subscriber<? super T> st = hook.onLift(operator).call(o);            try {                st.onStart();                parent.call(st);//调用上一级 OnSubscribe的 call() 方法            } catch (Throwable e) {                Exceptions.throwIfFatal(e);                st.onError(e);            }        } catch (Throwable e) {            Exceptions.throwIfFatal(e);            o.onError(e);        }    }}

（根据第一部分，On3包括两部分的内容： Func1 + On2）

结合源码和上图，可以看到 On3.call(Sb) 实际上分为两步：
a、生成新的Subscriber。新生成的Subscriber记为 Sb2，它包含两部分内容：Sb 和 On3 携带的 Func1_3
b、On3包含的On2 调用 call()方法,入参为新生成 Sb2.

3、On2.call(Sb2)
与上步骤类同，

4、On1.call(Sb1)
即对应示例代码中的以下片段：

由上图，所以下步就到了 Sb1.onNext(10)

5、Sb1.onNext()
注意：此处的 Sb1 实际上是 MapSubscriber 继承自 Subscriber

static final class MapSubscriber<T, R> extends Subscriber<T> {    final Subscriber<? super R> actual;//包含上一级的 Subscriber（对Sb1而言，即为 Sb2）    final Func1<? super T, ? extends R> mapper;//包含的 Func1（对Sb1而言，即为 Func1_3）    public MapSubscriber(Subscriber<? super R> actual, Func1<? super T, ? extends R> mapper) {        this.actual = actual;        this.mapper = mapper;    }    //onNext 操作    @Override    public void onNext(T t) {        R result;        try {            result = mapper.call(t);//调用包含的 Func1 的call方法        }         ...        actual.onNext(result);//调用包含的 Subscriber 的 onNext方法，入参为Func1执行后的结果    }    ...}

所以第一个map() 方法中的 Func1 被执行了，

6、Sb2.onNext()
同5，将 Func1_3 执行后的结果丢给下级 Subscriber

7、Sb.onNext()
Sb，即代码唯一可见的 Subscriber，现在到了最后一级 Subscriber

至此，调用完最后一级的 onNext()，整个代码流程结束

map() 解析

前面对map()只讲了结论，下面分析下原理。
对map() 源码从上至下截取代码片段，有助理解

//1、最顶层，传入 Func1.map(new Func1<Integer, Integer>() {     @Override     public Integer call(Integer integer) {         return integer *10;     } })

//2、调用了lift()方法，同时可以看到 Func1 被包在 OperatorMap 中（OperatorMap 继承自Operator）public final  Observable map(Func1<? super T, ? extends R> func) {    return lift(new OperatorMap(func));}

//3、返回一个新的Observable，原来的Observable的 onSubscribe 和 OperatorMap被包在了一个新的 OnSubscrbe中（OnSubscribeLift继承自 OnSubscribe）。public final  Observable lift(final Operator<? extends R, ? super T> operator) {    return new Observable(new OnSubscribeLift(onSubscribe, operator));}

综上，map()步骤与前面流程图对应，截图片段图如下：

即：
1、生成并返回新的Observable
2、新的Observable的 OnSubscribe 包含了上一级的 OnSubscribe 和 map传入的 Func1

参考资料

1、给 Android 开发者的 RxJava 详解

实例

步骤分析

map() 解析

参考资料

更多相关文章

随机推荐