亲宝软件园·资讯

展开

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

itbird01 人气:0

正文

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

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

订阅流

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

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

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

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

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

subscribe的解读收下

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

Observable是被观察者的意思,subscribe是订阅的意思,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
            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.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方法,而这个方法是个啥?在哪里实现的?一看,这玩意原来是Observable类中的一个抽象方法

protected abstract void subscribeActual(Observer&lt;? super T&gt; 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方法为开始,从右往左执行,这个执行过程中,每个节点,做两件事情

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

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

加载全部内容

相关教程
猜你喜欢
用户评论