RxJava构建流基本原理示例解析

时间:2023-02-03 itbird01 人气:0

正文

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

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

1.构建流

大家一定郁闷，怎么突然跑出来构建流的词汇，小编你之前也没讲到过呀？大家先别急，我们一步一步来。首先从一个最简单的，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");
                }
            });
}

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

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

这里运行一下，我们看到日志打印

大家看一下这里几个关键的函数对象

Observable 被观察者，subscribe订阅的意思，Observer观察者的意思，just是仅仅输入url，subscribeOn和observeOn是线程切换功能，map是中间的一个操作符。接下来，我们就这些看到的操作符、关键字，进行解读。

1.1 just的解读

从整体链式调用的代码形式上，我们大概可以猜测到，just、subscribeOn、observeOn、map等中间操作符，必定都是基于同一个对象的变形的builder模式，也就是说，他们应该都是同一个类内部的方法。

我们看一下的Observable的just源码

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

可以看到这里，仅仅是将传入的T 也就是我们上面样例中的urlString，包装为了ObservableJust对象。而just的调用，返回的依然是一个Observable被观察者对象。我们看一下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);
}

好了，just讲到这里，我们继续map，到目前为止，ObservableJust的subscribe方法仅仅是一个方法而已，并未调用到，所以我们暂时不予理睬，后面调用到，我们再去分析。

1.2 map的解读

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

从这个使用代码来看，map是just与subscribe之前的一个操作符，起到承上启下的作用。

我们看到首先是just返回的对象，可以直接调用map方法，我们之前分析、实现过just，知道just实际上返回的就是一个Observable，所以map方法就是Observable里面的一个方法。

再往下看，map返回后的对象，直接调用了subscribe方法，我们之前设计实现响应式功能代码的时候知道，subscribe是Observable里面的一个方法，那么这里也就知道了，map方法肯定也是返回一个Observable对象。

从上面分析，我们看到一个流向图

大家发现，这个流。只做了一件事情，每个操作符，都是对上一层的Observable进行了一层包装代理而已。这时我们提出一个概念，这个流，我们命名为构建流。

同理，subscribeOn、observeOn是否也是这样的呢？我们通过源码验证一下。

1.3 subscribeOn、observeOn

有了上面map的经验，我们猜测，subscribeOn、observeOn顺序调用的时候，应该也仅仅是对上一层的Observable进行了一层包装代理而已。我们看一下subscribeOn、observeOn的源码

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.CUSTOM)
    @NonNull
    public final Observable&lt;T&gt; subscribeOn(@NonNull Scheduler scheduler) {
        Objects.requireNonNull(scheduler, "scheduler is null");
        //果然这里包装为了ObservableSubscribeOn对象
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn&lt;&gt;(this, scheduler));
    }
    public final Observable&lt;T&gt; observeOn(@NonNull Scheduler scheduler, boolean delayError, int bufferSize) {
        Objects.requireNonNull(scheduler, "scheduler is null");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
        return RxJavaPlugins.onAssembly(new ObservableObserveOn&lt;&gt;(this, scheduler, delayError, bufferSize));
    }

我们继续看一下ObservableSubscribeOn、ObservableObserveOn类的源码

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }
    //...省略若干
}
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    final boolean delayError;
    final int bufferSize;
    public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
        super(source);
        this.scheduler = scheduler;
        this.delayError = delayError;
        this.bufferSize = bufferSize;
    }
}

而AbstractObservableWithUpstream本身就是继承与Observable

abstract class AbstractObservableWithUpstream<T, U> extends Observable<U> implements HasUpstreamObservableSource<T> {}

1.4 小结

从上面可以看到，也就是到目前为止，just、map、subscribeOn、observeOn一系列调用下来，依然都是执行一个操作而已，即每个操作符，都是对上一层的Observable进行了一层包装代理而已。

这就是构建流，构建流的作用就是，从开始操作符到订阅为止，从左往右执行，期间每个操作符，都是对上一层的Observable进行了一层包装代理而已。如图：

加载全部内容