RxJava 触发流基本原理源码解析

目录

• 正文
• 触发流
• 小结
• 总结

正文

本节，我们从RxJava使用代码入手，去结合自己已有的知识体系，加查阅部分源码验证的方式，来一起探索一下Rxjava实现的基本原理。

为了本文原理分析环节，可以被更多的人理解、学习，所以小编从初学者的角度，从使用入手，一点点的分析了其中的源码细节、思想，建议大家随着本文的章节步骤，一步一步的来阅读，才能更快、更好的理解Rxjava的真正的思想精髓，也为我们之后的实践课程留一个好的底子。

触发流

到目前为止，我们讲了构建流、订阅流，但是依然没有触发真正的observer中的事件，例如：

@Override
   public void onSubscribe(@NonNull Disposable d) {
       Log.d(TAG, "onSubscribe");
   }
   @Override
   public void onNext(@NonNull String s) {
       Log.d(TAG, "onNext s = " + s);
   }
   @Override
   public void onError(@NonNull Throwable e) {
       Log.d(TAG, "onError");
   }
   @Override
   public void onComplete() {
       Log.d(TAG, "onComplete");
   }

各位看官，莫急莫急，且听老衲娓娓道来。

还记得上面的订阅流吗？订阅流从右往左执行的，执行到最后的observable，执行了它的subscribe方法。我们从使用代码知道，最左端的observable是啥来着，大家还记得吗？当然是ObservableJust

private void test() {
	//第一步：just调用
    Observable.just("https://img-blog.csdn.net/20160903083319668")
    //第二步：map调用
            .map(new Function<String, Bitmap>() {
                @Override
                public Bitmap apply(String s) throws Exception {
                    //Bitmap bitmap编程客栈 = downloadImage(s);
                 开发者_开发培训   return null;
                }
            })
            //第三步：subscribeOn、observeOn调用
            .subscribeOn(Schedulers.newThread())
            .observeOn(androidSchedulers.mainThread())
            //第四步：subscribe调用
            .subscribe(new Observer<Bitmap>() {
                @Override
                public void onSubscribe() {
                    Log.d(TAG, "onSubscribe");
                }
                @Override
                public void onNext(Bitmap s) {
                    Log.d(TAG, "onNext s = " + s);
                }
                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "onError ", e);
                }
                @Override
                public void onComplete() {
                    Log.d(TAG, "onComplete");
                }
            });
}

我们就顺坡下驴，看一下ObservableJust的subscribe方法做啥了

public final class ObservableJust<T> extends Observable<T> implements ScalarSupplier<T> {
    private final T value;
    public ObservableJust(final T value) {
        this.value = value;
    }
    @Opythonverride
    protected void subscribeActual(Observer<? super T> observer) {
        ScalarDisposable<T> sd = new ScalarDisposable<>(observer, value);
        observer.onSubscribe(sd);
        sd.run();
    }
    @Override
    public T get() {
        return value;
    }
}

仔细一看，这里面没有subscribe方法，那么肯定就是调用父类observable的subscribe方法了

@SchedulerSupport(SchedulerSupport.NONE)
@Override
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;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Subscription already
        RxJavaPlugins.onError(e);
        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}

大家看到这里，其实关键在于，最终调用了一个subscribeActual方法，所以我们继续看子类ObservableJust的subscribeActual方法干啥了？

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        ScalarDisposable<T> sd = new ScalarDisposable<>(observer, value);
        observer.onSubscribe(sd);
        sd.run();
    }

接续根据ScalarDisposable的run方法

   public static final class ScalarDisposable<T>
    extends AtomicInteger
    implements QueueDisposable<T>, Runnable {
        private static final long serialVersionUID = 3880992722410194083L;
        final Observer<? super T> observer;
        final T value;
		//...省略很多代码
        @Override
        public void run() {
            if (get() == START && compareAndSet(STAAjmGVIzMRT, ON_NEXT)) {
            	//可以看到这里执行了onNext、onComplete方法
                observer.onNext(value);
                if (get() == ON_NEXT) {
                    lazySet(ON_COMPLETE);
                    observer.onComplete();
                }
            }
        }
    }

小结

看到这里，我们知道了，开始一层一层的从左往右去调用observer的相关方法了。 由订阅流可知，每层的observable实际上拥有下一层的observer的代理类，所以自然而然，从最左边开始调用observer的相关方法开始，触发流，就是从左往右，一层一层的剥开之前包裹的observer，然后顺序调用里面的onNext、onComplete等方法。 不信，我们挑一个ObservableMap来验证一下。

public finandroidal class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends U> function;
    public ObservableMap(ObservableSource<T> source, Function<? suhttp://www.devze.comper T, ? extends U> function) {
        super(source);
        this.function = function;
    }
    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));
    }
    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
        final Function<? super T, ? extends U> mapper;
        MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
            super(actual);
            this.mapper = mapper;
        }
        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            if (sourceMode != NONE) {
                downstream.onNext(null);
                return;
            }
            U v;
            try {
                v = Objects.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            //此处调用了下游的observer的onNext方法
            downstream.onNext(v);
        }
    }
}

可以看到里面，的确调用了下游的observer的onNext方法。

总结

整个过程，分为构建流、订阅流、触发流。

构建流：从左到右执行不同的操作符的过程，其实很简单，就是根据不同的操作符，对原始的 observable进行逐层包装，这里可以看出，每层的节点 N* 就持有了上一层的observable。

订阅流：从右到左的 subscribe 调用过程，这个过程中，每个observable内部的subscribeActual执行两个关键操作，一个是对自己已有的observer进行一层重新包装，另外一个就是使用前面节点的observable，订阅包装好的observer。

触发流：在订阅流执行完成之后，执行到最左端的observable，我们发现它内部的subscribeActual实现，实际上就是调用里面拥有的observer的相关回调方法（onNext、onComplete、onError等），那么这层回调流就简单了，就是一层一层的调用里面的observer，最终执行到最右端的observer。

篇幅所限，大家也发现了，我们本节课，我们详细讲解Rxjava线程切换的实现原理，这个有两个原因，一是篇幅所限，本节内容已经够多了，大家先吃透框架，另外一方面是，线程切换我相信我们后面实践环节，待框架自我搭建实现之后，里面的线程切换功能就是水到渠成的事情，相信凭借大家已有的知识，都可以做到的。

所以建议大家，先别看这块Rxjava是如何实现线程切换的，而是想一下，它是怎么实现的？到时我们自己的Rxjava框架搭建起来之后，填充实现一下。

提个醒儿，大家还记得我们之前EventBus源码分析、实践环节吗？其中也说到了线程切换。其实原理差不多，大家先想一下。

以上就是RxJava 触发流基本原理源码解析的详细内容，更多关于RxJava 触发流原理的资料请关注我们其它相关文章！