RxJava2 源码分析一

  • 时间:2019-06-11 03:47 作者:shenshizhong 来源:shenshizhong 阅读:20
  • 扫一扫,手机访问
摘要:文章目录前言RxJava2 详情RxJava2 使用带问题看源码总结前言在OkHttp3+Retrofit2+RxJava2 它是怎样在起作用的,为什么它会受到如此的欢迎,先通过分析如何发送数据,还有接收数据这块内容。本文是居于此版本分析的 implementation "io.reacti

文章目录

  • 前言
  • 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

  • 全部评论(0)
手机二维码手机访问领取大礼包
返回顶部