Android Rxjava 源码解析

前言

RxJava 这个名字，想必做过 Android 的人都有所了解，简单的使用几乎不成问题，但里面的知识点却一向令人望而生畏，接下来几日我便想慢慢解开这层面纱

先来回归一下简单的例子

没有线程切换的最简单的版本，初学者一定很困惑，这段代码是怎么运行起来的，这段代码中最具核心的是 Observable，Observable 有一个方法是 subcribe，当调用这个方法时便调用了 ObservableOnSubscribe.subscibe 方法，里面调用了 e.onNext()，也就是调用了 Obersver 的 onNext 方法，整个的流程便是这样，围绕这个流程，便衍生出了线程切换，各种类型的任务，背压。。。。。。

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext("hello");
                e.onNext("world");
                e.onComplete();

            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(String o) {
                System.out.println(o.toString());
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

map

为了方便理解，举一个简单的例子，实际的Rajava 与之类似
映射，将一个 Observable 转化成另一个 Observable 的过程，主要是 onNext 的调用链的关系，实际想起来还是挺复杂的，一个递归的结构，每一个 map 操作都会封装一个 Observable 而他的参数是 MapSubscribe 的发射器，里面在调用 onNext 时穿进去的参数是 transformer.call 的结果，这就说明了如果有 map 存在，首先会递归到开始，然后调用离之最近的 mapSubcriber 的 call 函数，获取结果，并调用下一个发射器的onNext函数，知道调用最顶层的Subcriber的回调函数，该过程还是很有意思的

Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                subscriber.onNext(10);
            }
        }).map(new Observable.Transformer<Integer, String>() {
            @Override
            public String call(Integer from) {
                return String.valueOf(from);
            }
        }).map(new Observable.Transformer<String, String>() {
            @Override
            public String call(String from) {
                return String.valueOf(from + 10);
            }
        }).map(new Observable.Transformer<String, String>() {
            @Override
            public String call(String from) {
                return String.valueOf(from + 10);
            }
        }).subscribeOn(Schedulers.io()).subscribe(new Subscriber<String>() {
            @Override
            public void onStart() {
                System.out.println("onStart called"+ " " + Thread.currentThread());
            }

            @Override
            public void onCompleted() {
                System.out.println("onComplete called"+ " " + Thread.currentThread());
            }

            @Override
            public void onNext(String object) {
                System.out.println(object + " " + Thread.currentThread());
            }

            @Override
            public void onError(Throwable t) {
            }
        });