RxJava实战之订阅流基本原理示例解析

目录

• 正文
• 订阅流
• subscribe的解读收下
• 小结

正文

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

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

订阅流

有人会问，小编，你到现在为止，只是讲了流程，而没有讲到具体每个中间操作符，在转换的对象里面的方法调用，这个问题，问的特别好！！！

还记得小编开篇说的那句话吗？我们从Rpythonxjava的使用代码入手

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");
                }
            });
}

从上面的样例代码分析、分解，我们明面上看到四个步骤，暂且列下来：

• 第一步：just调用
• 第二步：map调用
• 第三步：subscribeOn、observeOn调用
• 第四步：subscribe调用

之所以我们没有讲到ObservableObserveOn、ObservableMap、ObservableJust等对象里面的具体方法调用，是因为到目前为止，从使用例子入手，根本就没有调用到，所以我们也就无从分析到。，本节，接下来我们分析subscribe调用，大家就发现，里面的某些方法开始调用上了。

subscribe的解读收下

我们知道，上面的just、map、subscribeOn、observeOn一系列调用下来，依然是一个Observable对象、

Observable是被观察者的意思，subscwww.devze.comribe是订阅的意思，Observer是观察者的意思。

大家发现了没有？这里有个问题，这家伙和我们标准的观察者有很大的不同，标准观察者模式，是一种一对多的行为型设计模式，其实就是若干个观察者，将自身的接口引用注册到被观察者内部，被观察者状态发生变更时，遍历内部的list列表，一一通知观察者，如下图

Observable.just("https://img-blog.csdn.net/20160903083319668")
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.d(TAG, "onSubscribe");
            }
            @Override
       开发者_Go入门     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");
            }
        });

可是我们看上面Rxjava的代码，与标准观察者模式有两点不同

• 一是被观察者订阅了观察者
• 二是从使用上看观察者和被观察者的订阅关系是一对一的

上面提出的两点不同，我们一边看源码，一边试着去理解一下。

• 一对一的通知：因为响应式编程思想的重点在于，一个变化，另外一个要能感知到，那么通过这样变形的观察者模式，去实现一对一的通知，我觉得也没啥问题。
• 被观察者订阅观察者：这个从理论上讲，就没办法去理解了，对吧，因为你再怎么变形标准观察者模式，那也肯定是观察者订阅被观察者，所以这里我们有必要简单通过源码去了解一下

我们从上面看到just是将传入的T，再次封装为了一个ObservableJust对象

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> just(T item) {
    ObjectHelper.requireNonNull(item, "item is null");
    return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
}

我们看一下ObservableJust类代码

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

从上面看到，ObservableJust仅仅是将传入的T封装了一层而已，它继承与Observable抽象类，而Observable抽象类实现了ObservableSource接口

public abstract class Observable<T> implements ObservableSource<T> {

而ObservableSource接口，就是我们外界调用的subscribe订阅方法的源头

public interface ObservableSource<T> {
    /**
     * Subscribes the given Observer to this ObservableSource instance.
     * @param observer the Observer, not null
     * @throws NullPointerException if {@code observer} is null
     */
    void subscribe(@NonNull Observer<? super T> observer);
}

所以Observable肯定实现了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.require编程NonNull(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方法，而这个方法是个啥？在哪里实现的？一看，这玩意原来是Observable类中的一个抽象方法

protected abstract void subscribeActual(Observer<? super T> observer);

所以这里绕回到开头，我们知道just，实际上是将传入的参数T，转换封装为了ObservableJust对象，而ObservableJust继承与 Observable，所以subscribeActual方法它肯定去了

public final class ObservableJust<T> extends Observable<T> implements ScalarCallable<T> {
    private final T value;
    public ObservableJust(final T value) {
        this.value = value;
    }
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        ScalarDisposable<T> sd = new ScalarDisposable<T>(observer, value);
        //最终这里还是调用了观察者的相应方法
        observer.onSubscribe(sd);
        sd.run();
    }
    @Override
    public T call() {
        return value;
    }
}

小结

大家，发现了没有，这里绕了一圈，最终调用通过Observable的抽象方法subscribeActual的巧妙实现，最终还是观察者订阅了被观察者，被观察者内部最终调用了观察者的具体方法。

这里和标准观察者模式不同的是，被观察者立马去通知了观察者，说直接点，在调用被观察者的订阅方法时，其实就是直接调用了观察者相应的方法，只不过这里通过模板方法模式，巧妙的封装了，好了，Rxjava的观察者模式源码，我们简单理解到这里，我们试着自己去编写实现一下。

也就是订阅流的过程中，是以执行subscribe方法为开始，从右往左执行，这个执行过程中，每个节点，做两件事情

• 对后面的observer节点，做一层包装代理，变为代理的observerProxy

由于构建流的执行，每个节点实际上拥有上一个节点observable对象的引用，所以执行 source.subscribe(observerProxy)

订阅流讲到现在，大家是否理解了？当然这里没有详细讲解其中ObservableSubscribeOn、ObservableObserveOn中的订阅，如何进行的线程切换，这个并非是不去讲，还是那句老话，饭要一点一点的吃，我们congoing使用方法入手，想要去了解的是Rxjava的整体框架原理。至于线编程客栈程切换如何实现的？这个留个念想，大家可以认真想一下，不建议大家直接去看源码。我们在Rxjava实践环节，也会带大家一点一点的去实现这个核心功能。

以上就是RxJava实战之订阅流基本原理示例解析的详细内容，更多关于RxJava订阅流基本原理的资料请关注我们其它相关文章！