RxJava源码分析-数据流向

2018-05-21 本文已影响0人一只胖Wa牛

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，然后才执行了onNext、onComplete
接下来我们就来跟进代码看下整个流程是如何执行的

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源码分析-数据流向

RxJava源码分析-数据流向

Observable.create

RxJavaPlugins.onAssembly

ObservableCreate

Observable.subscrib

ObservableCreate.subscribeActual

猜你喜欢

热点阅读