拆轮子系列--RxJava理解（一）--Map解析

作者: 巨石强森小童鞋 | 来源:发表于2018-07-16 11:19 被阅读0次

本系列文章如下：

上一篇文章属于RxJava的前奏篇，大体介绍了下为什么要使用RxJava，本篇文章主要从RxJava的基本使用流程入手来对RxJava的源码进行分析，本文的分析源码基于RxJava1来进行分析，RxJava2的使用核心思想区别不大，对于理解不产生什么影响。本文的大纲如下：

RxJava的基本使用流程分析
RxJava操作符--map源码解析
总结

1. RxJava的基本使用流程分析

首先，先以一个RxJava最普通的使用方式进行开始：

Observable.OnSubscribe<String> onSubscriber1 = new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("17");
                subscriber.onCompleted();
            }
        };

        Subscriber<String> subscriber1 = new Subscriber<String>() {
            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(String s) {
            }
        };

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

上面的代码其实没有干什么事情，就是发射了一个String字符串。其意义在于，这个是RxJava最基本的一个订阅关系，而我们要分析的就是这个订阅关系是怎么工作的。
分析调用关系，最好就是查看源码，首先，我们看下create()这个方法的源码：

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

从create()方法中我们可以看出，其内部就是创建了一个被观察者，这里我们称为Observable1。查看Observable的构造函数：

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

在Observable的构造方法中，主要是做了两件事情：

1. 创建了一个observable1对象
2. 将我们传入的onSubscriber1保存在了observable1的onSubscribe属性中

接下里分析subscrible()方法，首先，先看看源码：

public final Subscription subscribe(Subscriber<? super T> subscriber) {
        return Observable.subscribe(subscriber, this);
    }

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

这里只是关注了几行核心代码，省略掉其余的代码。在subscribe()方法中，很明显的就是调用了hook.onSubscribeStart().call()这方法，接下来我们查看onSubscribeStart()方法：

public <T> OnSubscribe<T> su (Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {
        return onSubscribe;
    }

在onSubscribeStart()方法中，直接返回了observable1.onSubscribe。所以，subscribe()方法其实就是做了一件事情，就是调用我们栗子中的onSubscriber1的call()方法，而传入的对象就是栗子中的subscriber1，call()方法所执行的操作就是调用subscriber1对象的onNext()与onCompleted()方法。
我们用一张图总结上面方法的调用关系：

RxJava的create与subscribe.png

2. RxJava中常用操作符---map

在RxJava中，map操作符是一个很常用也很实用的操作符，现在我们主要分析下map操作的工作原理。首先，我们还是用一个例子进行说明：

Observable.OnSubscribe<String> onSubscriber1 = new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("17");
                subscriber.onCompleted();
            }
        };

        Subscriber<Integer> subscriber1 = new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(Integer result) {
             Log.w("TAG","output---" + result);
            }
        };

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

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

上面的例子增加了一个map的操作符，主要的作用就是将String类型的“17”转换成int型进行log输出。
上面的类型转换其实是map操作符很常用的一个功能，现在我们具体分析下这个操作符的底层是怎么实现的：

public final <R> Observable<R> map (Func1<? super T, ? extends R> func) {
        return lift(new OperatorMap<T, R>(func));
    }

首先还是走create()方法逻辑，创建了一个observable1，并且将传入的参数onSubscriber1保存在observable1的onSubscribe属性中，接下来我们重点关注下map操作符干了什么事情。
从源码中我们发现了两个新的东西，lift()与OperatorMap()。至于func1这个接口，它其实就是继承了Function，里面就只有一个call()方法。这里提及一下，在RxJava中会存在类似于Func+数字的接口，而这些接口中只有一个call()方法，类似的还有Action+数字的接口，它继承Action，里面页存在一个call()方法，与func不同的是，Action接口的call()方法都没有返回值，而Func接口的call()方法都存在返回值。这里我们将它们的源码贴出来进行对比：

/**
 * Represents a function with one argument.
 */
public interface Func1<T, R> extends Function {
    R call(T t);
}

/**
 * A one-argument action.
 */
public interface Action1<T> extends Action {
    void call(T t);
}

好了，言归正传，我们继续分析map。我们先看看OperatorMap到底做了什么操作：

public final class OperatorMap<T, R> implements Operator<R, T> {...}

从源码中我们看到OperatorMap实现了Operator，那么这个Operator到底是个什么东西？看下源码：

   /**
    * Operator function for lifting into an Observable.
    */
    public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<? super T>> {...}

好了，到这里也就明白了，Operator继承了Func1与Subscriber这个抽象类，前面提到过，func接口中只有一个带返回值的call()方法。也就是说，OperatorMap这个类其实就是重写了Func接口中的call()方法，至于做了什么事情，我们继续分析，先看下OperatorMap的构造函数：

public OperatorMap(Func1<? super T, ? extends R> transformer) {
        this.transformer = transformer;
    }

看过源码之后，发现这里其实啥也没干，就是将传入进来的transformer保存到OperatorMap中的属性中。等等，我们之前调用map()操作符的时候传入了一个transformer1，那么这个transformer1就是保存在OperatorMap的transformer属性中。接下来我们看看OperatorMap中重写了func接口的call()方法，到底做什么事情呢，继续查看源码：

@Override
public Subscriber<? super T> call(final Subscriber<? super R> o) {
        return new Subscriber<T>(o) {

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

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

            @Override
            public void onNext(T t) {
                try {
                    o.onNext(transformer.call(t));
                } catch (Throwable e) {
                    Exceptions.throwOrReport(e, this, t);
                }
            }
        };
    }

从源码中发现，这个call()方法就是接受一个subscriber类型的参数，然后返回另外一个Subscriber类型的对象。其实我们可以把OperatorMap看成一个订阅者，这样就可以监听Observable发出来的事件。在call()方法中，其实最重要的就是o.onNext(transformer.call(t));这个方法。
局部总结下OperatorMap中做的事情：

1. 重写了Func接口的call()方法，接收一个subscriber类型的参数，返回另一个subscriber类型的对象
1. 将传入的transformer1对象保存在operatorMap的属性中
1. 在重写的call()方法中，调用了transformer.call(t)，就是调用了transformer1中的Integer.parseInt(s)，简单的说，就是传入的subscriber调用了其onNext()方法，执行了Integer.parseInt(s)并返回了一个新创建的Subscriber对象

接下来，继续分析下map操作符中的另一个比较重要的东西--lift(...)。老样子看看源码：

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return new Observable<R>(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> o) {
                try {
                       Subscriber<? super T> st = hook.onLift(operator).call(o);
                    try {
                        st.onStart();
                        onSubscribe.call(st);
                    } 
                    ...
            }
        });
    }

截取了lift()中的重要代码，我们可以看到，lift()方法接收了一个Operator类型的参数，返回一个新创建的Observable类型对象。

3. 总结

分析了这么多，我们结合上面的例子，将map整个调用流程梳理一遍：

1. 首先，Observable.create(onSubscriber1)。这个里面做的操作就是创建了一个Observable对象observable1，并且将onSubscriber1对象保存在observable1对象的onSubscribe属性中
1. 执行了map(transformer1)操作。这里面的操作稍微复杂点。1) 先将transformer1对象保存在OperatorMap中的transformer属性中；2) 调用lift()方法，返回一个observable2；3) 在调用map()之后会调用subscribe()方法，也就是调用了observable2.subscribe()，同理，就是调用了observable2.onSubscribe.call()方法；4）call()方法首先做的事情就是传递一个subscriber1，返回一个subscribe对象，这里标记为subscribe_map；5）回到lift()方法继续执行，接着调用onSubscribe.call(st)，这里的st就是上一步返回的subscribe_map对象，也就是onSubscribe.call(subscribe_map); 6) onSubscribe对象就是我们第一步保存的onSubscriber1，所以现在的执行逻辑是onSubscriber1.call(subscribe_map)
1. 结合上面例子的情况，onSubscriber1.call(subscribe_map)的调用过程就是，先调用subscribe_map的call()，subscribe_map的call()操作会执行transformer1.call()，也就是执行了Integer.parseInt(“17”)，返回17; 接着调用onSubscriber1.call()，执行subscriber1.onNext()，输出17。
  结合例子的情况，整个调用过程我们用一个流程图来进行直观阐述：
  RxJava中map的调用.png
  到此，添加了map操作符的这个例子基本上梳理完毕了，map操作符只是RxJava中一个常用的操作符，RxJava有很多灵活的操作符，这里就不一一列举了，操作符的合理使用往往能提升我们的开发效率，让我们的程序变得简洁。

拆轮子系列--RxJava理解（一）--Map解析

1. RxJava的基本使用流程分析

2. RxJava中常用操作符---map

3. 总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读