RxJava2 源码分析一

文章目录

• 前言
• RxJava2 介绍
• RxJava2 使用
• 带问题看源码
• 总结

前言

在OkHttp3+Retrofit2+RxJava2 它是怎么在起作用的，为什么它会受到如此的欢迎，先通过分析如何发送数据，还有接收数据这块内容。
本文是居于此版本分析的

    implementation "io.reactivex.rxjava2:rxjava:2.2.8"
    implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

RxJava2 介绍

• 目前最热门的响应式函数编程框架
  RxJava Reactive Extensions for the JVM a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
  (JVM的反应性扩展:一个库，用于使用Java VM的可观察序列组合异步和基于事件的程序。）
  简单点就是一个异步+链式操作的库

RxJava2 使用

private void emitter() {
       Observable.create(new ObservableOnSubscribe<String>() {
           @Override
           public void subscribe(ObservableEmitter<String> emitter) throws Exception {
               emitter.onNext("1");
               emitter.onComplete();
           }
       }).subscribe(new Observer<String>() {
           @Override
           public void onSubscribe(Disposable d) {
               Log.e(TAG,"onSubscribe() d=" + d );
           }

           @Override
           public void onNext(String s) {
               Log.e(TAG,"onNext() s=" + s );
           }

           @Override
           public void onError(Throwable e) {
               Log.e(TAG,"onError() e=" + e );
           }

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

带问题看源码

从以上使用我们想知道以下两个问题
1、Observable 怎么发送数据
2、Observer 如何接收数据

• 通过以上的使用一步步来了解
  1、创建

 public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

然后继续看ObservableCreate

public final class ObservableCreate<T> extends Observable<T> {
       public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }
}

ObservableCreate 传入的参数 ObservableOnSubscribe 然后保存在this.source中，最终再返回 Observable，其实是ObservableCreate。（ObservableCreate 继承了Observable）

2、订阅
这是订阅的源码Observable #subscribe

public final void subscribe(Observer<? super T> observer) {
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            observer = RxJavaPlugins.onSubscribe(this, observer);

            ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

            subscribeActual(observer);  //这是关键部分，是去调用真实的订阅
        } catch (NullPointerException e) { // NOPMD
            throw e;
      ...
        
        }
    }

subscribeActual(observer); 这是订阅的主要过程，那么是谁实现的呢，其实就是上面我们创建后返回的ObservableCreate，这个时候我们继续看 ObservableCreate 中的 subscribeActual

真正订阅过程ObservableCreate#subscribeActual

protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);   
        try {
            source.subscribe(parent);  //订阅最关键代码
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }
}

结合创建订阅的过程，我们知道 observer其实就是new Observer<String>，source 就是 ObservableOnSubscribe，parent 就是CreateEmitter。

source.subscribe(parent); 其实就是ObservableOnSubscribe 调用subscribe中的CreateEmitter。CreateEmitter是怎么实现的呢？

这个时候我们去看CreateEmitter的实现

 CreateEmitter(Observer<? super T> observer) {
            this.observer = observer;
        }

        @Override
        public void onNext(T t) {
.....
            if (!isDisposed()) {
                observer.onNext(t);   //回调new Observer那个onNext方法
            }
        }

        @Override
        public void onError(Throwable t) {
            if (!tryOnError(t)) {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public boolean tryOnError(Throwable t) {
...
            if (!isDisposed()) {
                try {
                    observer.onError(t);//回调new Observer那个onError方法
                } finally {
                    dispose();
                }
                return true;
            }
            return false;
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();//回调new Observer那个onComplete方法
                } finally {
                    dispose();
                }
            }
        }

代码中有相关的注释，以上代码最主要在说明，为什么我们的回调方法中的代码会被执行，并且可以接收到数据，也就是在CreateEmitter 中调用了，而 CreateEmitter 是在ObservableOnSubscribe 的 subscribe(ObservableEmitter<String> emitter) 中执行的，就是我们写的 emitter.onNext("1");进行了数据的发送

• 过程总结：
  1、创建 Observable 返回 ObservableCreate
  2、ObservableCreate 进行 subscribe，然后执行 subscribeActual
  3、在 subscribeActual 中实际上是 ObservableOnSubscribe 调用 subscribe 然后执行 emitter.onNext("1");//发送
  4、emitter.onNext("1"); 最终会去调用new Observer 中的 onNext（）
  5、最终我们写在onNext() 的代码就被执行了，并且可以获取传过来的数据//接收

总结

• 是不是看到了RxJava2 的神秘面纱，发现原来这么好玩
• 主要还是发送接收过程中巧妙的使用抽象类，并做好回调工作
• 后续还会分析线程之间的调度，一点点感受RxJava2的美妙

如果对你有一点点帮助，那是值得高兴的事情。：)

我的csdn：http://blog.csdn.net/shenshizhong
我的简书：http://www.jianshu.com/u/345daf0211ad