RxJava源码分析之变换

RxJava 简单来说 , 是一个很灵活切换线程的裤子 .

• 简单试例
• 源码解读试例
• 变换思想图解
• 变换思想总结

1 、 简单试例

    
    Observable.just("url")//
        .map(new Function<String, Long>() {
            @Override
            public Long apply(@NonNull String s) throws Exception {
                return 1l;
            }
        })

        .map(new Function<Long, Integer>() {
            @Override
            public Integer apply(@NonNull Long l) throws Exception {
                return 50;
            }
        })
         .subscribe(new Observer<Integer>() {
             @Override
             public void onSubscribe(@NonNull Disposable d) {

             }

             @Override
             public void onNext(@NonNull Integer integer) {

                 Log.i("Info","onNext:"+integer);

             }

             @Override
             public void onError(@NonNull Throwable e) {

                 Log.i("Info","异常:"+e.getLocalizedMessage());
                 e.printStackTrace();
             }

             @Override
             public void onComplete() {

             }
         });
         
         打印:nNext:50
    

2 、 源码解读实例

创建 ObservableJust 对象

    
    public static <T> Observable<T> just(T item) {
        ObjectHelper.requireNonNull(item, "The item is null");
         //这里其实是直接返回ObservableJust对象 . 
        return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
    }
    
     public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
            //默认情况下 onObservableAssembly 为空 , 需要全局传入
        Function<? super Observable, ? extends Observable> f = onObservableAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;

    }

RxJavaPlugins.onAssembly 这里起到了 Hook 作用 , 加多了一层 , 可以用于 Fliter 例如

RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {


            @Override
            public Observable apply(@NonNull Observable observable) throws Exception {
                     Log.i("Info","Hook  创建对象 Observable ");
                if(observable instanceof ObservableJust && !((((ObservableJust)(observable)).call()) instanceof String) ){

                    return null;
                }
                return observable;
            }
        });
        
        //再一次运行上面的例子就是打印出:Hook  创建对象 Observable 

使得所有属于全局 ObservableJust 的 value 非 String 类型都无法正常创建 . 下面进入重点变换
变换也是一样加了一层全局 Hook . 这里直接默认返回 new ObservableMap<T, R>(this, mapper)即可

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
    }
    

ObservableMap 也是 Observable 的一个子类 , 把 ObservableJust 和 变换函数 Function 传进去创建 ObservableMap , 下一个 ObservableMap 走相同逻辑 , 这样就形成 ObservableJust(1) ∈ ObservableMap(2)
∈ ObservableMap(3) 的关系 (∈属于) 典型的责任链模式 . 上面例子逻辑走到 ···(A).subscribe(new Observer<Integer>()··· (A)处的对象为ObservableMap(3) , 发生订阅 . 继续查看源码

public final void subscribe(Observer<? super T> observer) {
                        ···· 省略N多逻辑
                    subscribeActual(observer);

subscribeActual是一个抽象方法 , 查看子类 ObservableMap 实现该方法

public void subscribeActual(Observer<? super U> t) {
            //这里的source为ObservableMap(2)对象
        source.subscribe(new MapObserver<T, U>(t, function));

    }

source 调用 subscribe 跟 ObservableMap(3) 执行相同逻辑 , 这里的 source 指的是 ObservableMap(2) 对象 , 这样就形成了(有序序列). MainActivity 里的 内部类 Observer 被包装了一层 MapObserver 当成参数往上传形成了链表 MapObserver(3) --> MapObserver(2) --> Observer(1) .

最终回到 ObservableJust 发生订阅把数据源于流的形式一层一层发射 , 见

//ObservableJust 的subscribeActual 方法
protected void subscribeActual(Observer<? super T> s) {
        Log.i("Info","subscribeActual 发射 ");
        //这里的s 是MapObserver(3)对象
        ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
        //observer 的 onSubscribe 方法被回调
        s.onSubscribe(sd);
        sd.run();

    }
    
    public void run() {
            if (get() == START && compareAndSet(START, ON_NEXT)) {
                    //正式发射数据源
                observer.onNext(value);
                if (get() == ON_NEXT) {
                    lazySet(ON_COMPLETE);
                    observer.onComplete();
                }
            }
        }

数据源发射 MapObserver(3).onNext() --(Function变换)--> MapObserver(2).onNext() --(Function 变换)-->Observer(1).onNext() .

3 、 变换思想图解

p.png

说明 : subscribe 是发生在 Observable 里面调用subscribeActual , 内部创建 MapObserver 包装对象 Observer , 传给父级Observable . source.subscribe(new MapObserver<T, U>(t, function));

4、 变换思想总结.

变换是 RxJava 最核心思想 , RxJava 源码比较绕 , 但是万变不离其中 , 包括线程调度 subscribeOn() 和 observeOn() 也是利用变换 .