美文网首页
RxJava源码分析-数据流向

RxJava源码分析-数据流向

作者: 一只胖Wa牛 | 来源:发表于2018-05-21 18:40 被阅读0次

RxJava源码分析-数据流向

  • RxJava使用也有很长一段时间了,RxJava说实话理解很难,但是使用很简单,源码其实也很精妙,本文分析基于Rxjava2.0
  • 推进几个好文,给 Android 开发者的 RxJava 详解
  • 给初学者的RxJava2.0教程作者比作水管的理解很有意思
  • 分析之前首先我们先提几个问题
    1.数据是如何创建封装成Observable的?
    2.数据是如何从上游的Observable发送到Observer的?
    先上一段代码
Observable.create(ObservableOnSubscribe<String> {
            it.onNext("hello,world")
            it.onComplete()
        }).subscribe(object : Observer<String> {

            var mDisposable: Disposable? = null
            override fun onComplete() {
                Log.d("Observer", "onComplete")
            }

            override fun onSubscribe(d: Disposable) {
                mDisposable = d
                Log.d("Observer", "onSubscribe")
            }

            override fun onNext(t: String) {
                Log.d("Observer", "onNext:-- $t")
            }

            override fun onError(e: Throwable) {
                Log.d("Observer", "onError:--")
            }

        })

我们来看logcat的输出

05-21 09:40:09.230 25346-25346/com.example.pandaguo.rxdemo D/Observer: onSubscribe
    onNext:-- hello,world
    onComplete

先后打印了,也就说下游的Observer代码执行顺序是先执行了onSubscribe,然后才执行了onNextonComplete
接下来我们就来跟进代码看下整个流程是如何执行的

Observable.create

我们的示例代码中,创建了一个ObservableOnSubscribe对象并将其作为参数传入到了Observable的静态方法create中,我们就看下整个方法做了什么

public abstract class Observable<T> implements ObservableSource<T> {
    ...
    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        //null校验
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }
    ...
}

可以看到这里就是简单的做了一个非空检验,然后就执行了RxJavaPlugins的onAssembly方法,并new了一个ObservableCreate对象作为参数传入,继续跟进

RxJavaPlugins.onAssembly

public final class RxJavaPlugins {
    ...
    static volatile Function<? super Observable, ? extends Observable> onObservableAssembly;
    ...
    public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
        Function<? super Observable, ? extends Observable> f = onObservableAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }
    ...
}

首次执行RxJavaPlugins的onAssembly方法时,此时onObservableAssembly还未被初始化,所以就直接返回了传入的source,也就是前一步我们直接创建的ObservableCreate对象,所以可以这么说Observable.create()最终我们得到的就是一个ObservableCreate对象,我们来看下ObservableCreate的实现

ObservableCreate

public final class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }
    ...
}
  • 整个类的实现代码比较多我们一点点的看
  • 首先,ObservableCreate继承自Observable
  • 构造方法中,其实就是将Observable.create()方法中我们自己创建的ObservableOnSubscribe匿名对象赋值给了成员变量source,如此ObservableCreate就持有了数据源
  • 到此我们知道了Observable对象时怎么创建并包装成ObservableCreate对象的
    那么还有一个问题,上游的Observable时怎么把数据发送给下游的Observer呢?我们接着看示例代码,紧接着会用调用到Observable.subscrib(),并且我们创建了一个Observer对象作为参数传入,跟进查看代码

Observable.subscrib

public abstract class Observable<T> implements ObservableSource<T> {
    ...
    try {
        ...
        subscribeActual(observer);
    } catch (NullPointerException e) {
        ...
    } catch (Throwable e) {
        ...
    }
    ...
}

略去不必要的代码,实际上这个方法最终调用到了subscribeActual方法,在Observable中subscribeActual方法是个抽象方法,因此它的具体实现就是在ObservableCreate对象中,我们来看下ObservableCreate中都做了什么

ObservableCreate.subscribeActual

public final class ObservableCreate<T> extends Observable<T> {
    ...
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);
        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            ...
        }
    }
    ...
}
  • 这个方法中我们可以看到首先是创建了一个CreateEmitter对象并把observer作为参数传入,
  • 之后回调了observer.onSubscribe(parent),由此可见为什么前面我们打印log时候onSubscribe是最先执行的
  • 最后也是最关键的,调用前面传入的ObservableOnSubscribe对象source的subscribe,并将第一步创建的CreateEmitter对象传入,这样子就关联起来上游的Observable对象与下游的Observer对象
public interface ObservableOnSubscribe<T> {
    void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}
  • 我们来回忆下前面在ObservableOnSubscribe的实现中,我们写了什么
    it.onNext("hello,world")
    it.onComplete()

这里的it实际上就是传入的CreateEmitter,那么CreateEmitter是怎么把数据传给Observer呢?

  • 我们来看下CreateEmitter的实现
public final class ObservableCreate<T> extends Observable<T> {
    ...
    static final class CreateEmitter<T>
    extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {

        private static final long serialVersionUID = -3434801548987643227L;

        final Observer<? super T> observer;

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

        @Override
        public void onNext(T t) {
            if (t == null) {
                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
                return;
            }
            if (!isDisposed()) {
                observer.onNext(t);
            }
        }

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

        @Override
        public boolean tryOnError(Throwable t) {
            if (t == null) {
                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
            }
            if (!isDisposed()) {
                try {
                    observer.onError(t);
                } finally {
                    dispose();
                }
                return true;
            }
            return false;
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

        @Override
        public void setDisposable(Disposable d) {
            DisposableHelper.set(this, d);
        }

        @Override
        public void setCancellable(Cancellable c) {
            setDisposable(new CancellableDisposable(c));
        }

        @Override
        public ObservableEmitter<T> serialize() {
            return new SerializedEmitter<T>(this);
        }

        @Override
        public void dispose() {
            DisposableHelper.dispose(this);
        }

        @Override
        public boolean isDisposed() {
            return DisposableHelper.isDisposed(get());
        }
    }
    ...
}
  • CreateEmitter是ObservableCreate的一个静态内部类,它继承自AtomicReference因此,他是线程安全的
  • 实现了ObservableEmitter、Disposable这两个接口
  • 首先我们来看下它的onNext方法,首先校验数据,之后调用isDisposed()判断是否已经消费掉了,如果说没有消费掉就调用Observer.onNext(),这样数据就传递给了Observer去处理
  • onComplete()、onError()不再去仔细分析了,唯一不同的是他会在调用Observer对应方法时候去调用dispose()方法表示已经消费掉了,这也就是为什么onComplete()、onError()其中只会有一个能执行,并且执行其中任意一个以后,onNext()就不会收到回调了
  • 至此整个流程就简单的分析完毕了
  • 下一篇来分析下RxJava的线程切换原理,欢迎看客大佬们来点个赞

相关文章

网友评论

      本文标题:RxJava源码分析-数据流向

      本文链接:https://www.haomeiwen.com/subject/axuqjftx.html