RxJava基本原理解析(一)

作者: yotods | 来源:发表于2019-03-21 14:33 被阅读0次

最近有时间，研究了一下Rxjava的实现，发现还是挺巧妙的一种架构(我是写不出来)
尝试的实现了一下，你猜怎么着～哈哈写出来了！(当然是最基本的)
其实不难，主要是绕。我这几篇写一下我的经验吧～

RxJava基本原理解析(一) - 主要基本理论
RxJava基本原理解析(二) - 线程切换之上行：subscribeOn (怎么办..懒得写)
RxJava基本原理解析(三) - 线程切换之下行：observeOn (怎么办..懒得写)
RxJava基本原理解析(三) - 操作符之：map (怎么办..懒得写)

一、链式编程

主要是说，rxjava中类似 rxjava.a().b().c();是怎么实现的，这点很重要，一定要理解。
我们看一下简单的Demo，不直接上Rx是为了更好的理解，后续会结合Rx一起看，事半功倍。
先写4个类，分别是A，A1，A2，A3。看代码

// 这是一个抽象类，A1,A2,A3都继承此类，并且实现了go的方法
public abstract class A {
    public abstract void go();
    // 创建A1
    public static A createA1() {
        return new A1();
    }
    // 创建A2
    public A createA2() {
        return new A2(this);
    }
    // 创建A3
    public A createA3() {
        return new A3(this);
    }
}

public class A1 extends A {
    // 实现go方法，打印a1
    @Override
    public void go() {
        Log.i("aaa", "a1");
    }
}

public class A2 extends A {
    private A a;
    //传入A对象
    public A2(A a) {
        this.a = a;
    }
    // 实现go方法，打印a2,并调用a.go
    @Override
    public void go() {
        Log.i("aaa", "a2");
        a.go();
    }
}

public class A3 extends A {
    private A a;
    //传入A对象
    public A3(A a) {
        this.a = a;
    }
    // 实现go方法，打印a2,并调用a.go
    @Override
    public void go() {
        Log.i("aaa", "a3");
        a.go();
    }
}

此时调用A.createA1().createA2().createA3().go();
如我们所愿：

I/aaa: a3
I/aaa: a2
I/aaa: a1

是怎么倒序打印出来的呢？
我们看到在createA2和createA3时，分别传入了this，也就是A的实例。分析一下流程：

A.createA1() //此时 A=A1
 .createA2(); // new之前this=A=A1，所以在new A2(this)，是把A1传入了A2。new之后A=A2
 .createA3(); // new之前this=A=A2，所以在new A3(this)，是把A2传入了A3。new之后A=A3
 .go();

由于此时A=A3，所以先调用A3的go()，打印"a3"，A3中的A=A2，
所以a.go()，会调用A2的go()，打印"a2"，A2中的A=A1，
所以会调用A1的go()，打印"a1"。

说完了，下一个

二、接口互调

private interface IA {
    void subscribe(IB ib);
}

private interface IB {
    void onNext(String aa);

    void onError();

    void onComplete();
}

// 执行这个方法
private void simpleRxJava() {
    IA ia = new IA() {
        @Override
        public void subscribe(IB ib) {
            ib.onNext("aaaa");
            ib.onError();
            ib.onComplete();
        }
    };

    IB ib = new IB() {
        @Override
        public void onNext(String aa) {
            Log.i(TAG, aa);
        }

        @Override
        public void onError() {
        }

        @Override
        public void onComplete() {
        }
    };
    ia.subscribe(ib);
}

看完这个，是不是觉得有点眼熟，哦不，是很眼熟。
没错，你已经掌握了RxJava的精髓！！
此时，你有想过把 链式编程+接口互调 结合在一起吗？？？如果没有，请右上角叉叉离开这里吧！

三、结合RxJava

之后提到一下这段代码，我就把它称为 经典代码 吧。

// 经典代码
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("经典代码");
    }
})
        // .subscribeOn(Schedulers.io()) // 这篇文章不说这个
        // .observeOn(AndroidSchedulers.mainThread()) // 这篇文章不说这个
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(String s) {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

看着有没有很像接口互调的例子呢？我们看一下ObservableOnSubscribe和Observer。

// ObservableOnSubscribe
public interface ObservableOnSubscribe<T> {
    void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}

// Observer
public interface Observer<T> {
    void onSubscribe(@NonNull Disposable d);
    void onNext(@NonNull T t);
    void onError(@NonNull Throwable e);
    void onComplete();
}

这两个都是接口，但是void subscribe(@NonNull ObservableEmitter<T> emitter) ObservableEmitter是什么，为什么不是Observer呢？我们来看一下ObservableEmitter。

// ObservableEmitter
public interface ObservableEmitter<T> extends Emitter<T> {
    void setDisposable(@Nullable Disposable d);
    void setCancellable(@Nullable Cancellable c);
    ObservableEmitter<T> serialize();
    boolean tryOnError(@NonNull Throwable t);
}

// Emitter
public interface Emitter<T> {
    void onNext(@NonNull T value);
    void onError(@NonNull Throwable error);
    void onComplete();
}

我们发现了ObservableEmitter extends Emitter Emitter中有我们想要的那三个方法，那么onSubscribe()怎么办呢？它们两个接口又是怎么连接起来的呢？然后又是怎么调用的呢？Observer怎么又变成了ObservableEmitter？
这一系列的问题，这是人性的扭曲，还是道德的沦丧，让我们跟create()方法走进Rxjava的内心世界。
哈哈哈哈哈哈哈

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

这时候就是rx的套路了，看到这里你只需要看new ObservableCreate<T>(source)就可以了。ObjectHelper.requireNonNull(source, "source is null");明显是一个判空；
RxJavaPlugins.onAssembly在方法里有Function接口，其实这个是用来给map操作符，类型转换用的，目前来说就是传什么返回什么，不予理会。
所以最终的代码到了new ObservableCreate(source)，要注意三个地方：
1.当前类是Observable，Observable是抽象类，并且实现了ObservableSource接口；
2.这个方法的返回类型仍然是Observable，连泛型T都没有变；
3.source是经典代码中 create(new ObservableOnSubscribe(){}) 实现的内部类，传入了ObservableCreate中。

// Observable implements ObservableSource
public interface ObservableSource<T> {
    void subscribe(@NonNull Observer<? super T> observer);
}

public final class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);

        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }
}

ObservableCreate是继承Observable的，看到这里，我们是不是有觉得和**链式编程**有点像，此时Observable=ObservableCreate，在ObservableCreate构造中，拿到了经典代码的ObservableOnSubscribe。还有一个方法subscribeActual(Observer<? super T> observer)是什么呢？

// Observable中的方法
protected abstract void subscribeActual(Observer<? super T> observer);

是个抽象方法，所以继承与Observable的所有子类，都需要实现这个方法，那么这个方法在哪调用的呢？

我们看一下经典代码的subscribe方法：

// Observable中的方法
public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);

        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

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

按照Rx套路，subscribe(Observer<? super T> observer) 这个observer就是我们onNext()...的回调，判空，同类型转换，判空，到了关键代码 subscribeActual(observer);，结合上面create，我们知道了，此时的Observable=ObservableCreate，这里调用了subscribeActual并且传入了observer，那么意味着实际调用在ObservableCreate -- subscribeActual里面，所以我们要看ObservableCreate#subscribeActual的实现

// ObservableCreate#subscribeActual
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);

source.subscribe(parent);

代码就三行：
第一行：创建了CreateEmitter，new的时候还把Observer回调传了进去；
第二行：把parent传了出去，回调到了经典代码的onSubscribe(Disposable d)方法；
第三行：接口互调，source是new ObservableCreate时带来的ObservableOnSubscribe，parent则是CreateEmitter。CreateEmitter
我们看一下

// ObservableCreate类里
// 实现了ObservableEmitter，和Disposable，所以在上面第二行observer.onSubscribe(parent);时没有问题
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {


    private static final long serialVersionUID = -3434801548987643227L;
    // 我是回调
    final Observer<? super T> observer;

    CreateEmitter(Observer<? super T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        if (t == null) {
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        if (!isDisposed()) {
            // 回调onNext
            observer.onNext(t);
        }
    }

    @Override
    public void onError(Throwable t) {
        if (!tryOnError(t)) {
            RxJavaPlugins.onError(t);
        }
    }

    @Override
    public boolean tryOnError(Throwable t) {
        if (t == null) {
            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
        }
        if (!isDisposed()) {
            try {
                // 回调onError
                observer.onError(t);
            } finally {
                dispose();
            }
            return true;
        }
        return false;
    }

    @Override
    public void onComplete() {
        if (!isDisposed()) {
            try {
                // 回调onComplete
                observer.onComplete();
            } finally {
                dispose();
            }
        }
    }

    @Override
    public void setDisposable(Disposable d) {
        DisposableHelper.set(this, d);
    }

    @Override
    public void setCancellable(Cancellable c) {
        setDisposable(new CancellableDisposable(c));
    }

    @Override
    public ObservableEmitter<T> serialize() {
        return new SerializedEmitter<T>(this);
    }

    @Override
    public void dispose() {
        DisposableHelper.dispose(this);
    }

    @Override
    public boolean isDisposed() {
        return DisposableHelper.isDisposed(get());
    }
}

我们看到CreateEmitter实现了ObservableEmitter和Disposable，并且持有Observer回调对象，所以在onNext，tryOnError，onComplete加了逻辑判断后，都是用Observer回调出去，也就是说调到了经典代码的回调中。
onError很特殊，是rx自己处理error时用的，给经典代码回调error则是tryOnError方法。
实现了Disposable，是为了通过第二行代码observer.onSubscribe(parent);回调到经典代码，来控制随时结束。

四、结语

整体流程为:

经典代码调用Observable.create创建了ObservableCreate，并把ObservableOnSubscribe传了进去，ObservableCreate是继承Observable的，并且实现了subscribeActual的方法；
经典代码调用subscribe后，此时Observable=ObservableCreate，实际上调用了ObservableCreate中subscribeActual方法，并且把回调接口Observer传入了ObservableCreate中。
所以ObservableCreate同时持有ObservableOnSubscribe和Observer
在ObservableCreate的subscribeActual中创建了一个中介类CreateEmitter，observer.onSubscribe(parent) 把Disposable回调到经典代码，调用了终极方法source.subscribe(parent)进行接口互调，回调到经典代码中create创建出的内部类，此时parent=CreateEmitter=ObservableEmitter；
当调用emitter.onNext("经典代码") 时，其实就调到了CreateEmitter中的onNext，然后通过observer.onNext(t) 回调到经典代码中最终回调的onNext方法。

就这样，接口之间来回调用，ObservableCreate用来连接两个接口，最终完成了Rxjava经典代码的回调。

RxJava基本原理解析(一)

一、链式编程

说完了，下一个

二、接口互调

三、结合RxJava

四、结语

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读