是什么让我们阅读RxJava源码如此艰难？

Posted 2021-04-30 启舰杂谈

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了是什么让我们阅读RxJava源码如此艰难？相关的知识，希望对你有一定的参考价值。

点击上方“ 启舰杂谈 ”，马上关注，

每天早上 8:30 准时推送

文章转自链接：https://juejin.im/post/5cce6fb05188254177317fdc

这是一篇需要用心去感受的文章，快读的阅读可能效果不好哟。

概述

你是不是看过了很多分析Rxjava源码的文章，但依旧无法在心中勾勒出Rxjava原理的样貌。

是什么让我们阅读Rxjava源码变得如此艰难？

是Rxjava的代码封装，以及各种细节问题的解决。

本文我把Rxjava的各种封装、抽象统统剥去，只专注于基本的事件变换。在理解了事件变换大概是做了件什么事情时，再去看源码，考虑一些其它问题就会更加容易。说明：这是一篇Rxjava源码分析的入门文章。旨在让读者脑中有个概念Rxjava最主要干了件什么事情，几个常用操作符的主要原理。今后再去看其它源码分析文章或源码能够更容易理解。因此本文先不去考虑Rxjava源码中复杂的抽象封装，线程间通信，onComplete、onError、dispose等方法，仅专注于“onNext”的最基本调用方式。

项目源码

https://github.com/OliverY/RxjavaYxj

本文目录：

手写Rxjava核心代码，create，nullMap（核心）操作符
map，observeOn，subscribeOn，flatMap操作符
响应式编程思想的理解

手写Rxjava核心代码，create，nullMap操作符

Create操作符

我们先来看一个最简单调用

MainActivity.java

Observable.create(new Observable<String>() {
    @Override
    public void subscribe(Observer<String> observer) {
        observer.onNext("hello");
        observer.onNext("world");
        observer.onComplete();
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onNext(String s) {
        Log.e("yxj",s);
    }

    @Override
    public void onComplete() {
        Log.e("yxj","onComplete");
    }
});



 
   
   
 
  
    
    
  Observable.java

public abstract class Observable<T> {

    public abstract void subscribe(Observer<T> observer);

    public static <T> Observable<T> create(Observable<T> observable){
        return observable;
    }

}       

 
   
   
 


 
   
   
 
  
    
    
  Observer.java

public interface Observer<T> {

    void onNext(T t);
    void onComplete();
}

 
   
   
 

本篇文章我把Observable称为“节点”，Observer称为“处理者”，一是因为我被观察者、被观察者、谁订阅谁给绕晕了，更重要的是我觉得这个名称比较符合Rxjava的设计思想。

Observable调用create方法创建一个自己，重写subscribe方法说：如果我有一个处理者Observer，我就把“hello”，“world”交给它处理。

Observable调用了subscribe方法，真的找到了Observer。于是兑现承诺，完成整个调用逻辑。

这里是“如果”有处理者，需要subscribe方法被调用时，“如果”才成立。Rxjava就是建立在一系列的“如果”（回调）操作上的。

“nullMap”操作符（核心）

1.创建一个observable
2.调用空map操作符做变换
3.交给observer处理


 
   
   
 
  
    
    
  MainActivity.java

Observable.create(new Observable<String>() {
            @Override
            public void subscribe(Observer<String> observer) {
                observer.onNext("hello");
                observer.onNext("world");
                observer.onComplete();
            }
        })
        .nullMap()
        .subscribe(new Observer<String>() {
            @Override
            public void onNext(String s) {
                Log.e("yxj",s);
            }

            @Override
            public void onComplete() {
                Log.e("yxj","onComplete");
            }
        });

nullMap()等价于下面这段代码。

即把上个节点的数据不做任何修改的传递给下一节点的map操作

.map(new Function<String, String>() {
    @Override
    public String apply(String s) throws Exception {
        return s;
    }
})

"nullMap"操作符在Rxjava源码里并不存在，是我方便大家理解Rxjava运行机制写出来的。

因为nullMap操作是一个 base变换操作，map，flatMap，subscribeOn，observeOn操作符都是在nullMap上修改而来。所以Rxjava的变换的基础就是nullMap操作符。

Observable.java
// 这就是Rxjava的变换核心

public Observable<T> nullMap() {

        return new Observable<T>() {
            @Override
            public void subscribe(final Observer<T> observerC) {

                Observer<T> observerB = new Observer<T>() {
                    @Override
                    public void onNext(T t) {
                        observerC.onNext(t);
                    }

                    @Override
                    public void onComplete() {
                        observerC.onComplete();
                    }
                };
                Observable.this.subscribe(observerB);
            }
        };
    }

“nullMap”操作符做了件什么事情：

上一个节点Observable A调用nullMap(),在内部new一个新的节点Observable B。
节点B重写subscribe方法，说"如果"自己有操作者Observer C，就new一个操作者Observer B，然后让节点A subscribe 操作者B。
节点A subscribe 操作者B，让操作者B执行onNext方法。操作者B的onNext方法内部，调用了操作者C的onNext。从而完成了整个调用。

请注意2中的”如果“。意味着，当节点B中的subscribe方法没有被调用的时候，2，3步骤都不会执行（他们都是回调），没有Observer B，节点A也不会调用subscribe方法。

接下来分两种情况：

节点B调用了subscribe方法，则执行2，3，完成整个流程。
节点B调用nullMap，从新走一遍1，2，3步骤，相当于节点B把任务交给了下一个节点C。

概况一下就是：

Observable每调用一次操作符，其实就是创建一个新的Observable。新Observable内部通过subscribe方法“逆向的”与上一Observable关联。在新Observable中的new出来的Observer内的onNext方法中做了和下一个Observer之间的关联。

图文详细解说nullMap整体调用过程

第一阶段

上图代表节点还未最终subscribe一个Observer，图中

步骤1：节点A调用map方法，在内部创建了一个新的节点B

这一阶段：主要就是节点与节点之间做连接，之间有各种“如果”（回调）的承诺。节点B这时候对节点A做了个承诺：“如果”我有处理者Observer C，那我就内部new一个 Observer B给你（节点A）用”。节点B中的操作者Observer B内部做了与Observer C的衔接工作。