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:  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.songma.com/u/345daf0211ad