一些使用RxJava的想法
在之前的项目中使用过 RxJava 配合MVP来搭建项目框架。当时刚开始接触RxJava，算是边学习边搭建项目。使用程度不深，代码思路和之前的项目代码思路不一样。链式调用我是比较喜欢的，但是感觉RxJava会不会太重？不过每个项目有每个项目的实际业务场景，感觉不是每个项目都适合使用RxJava，现在回想当时的项目使用RxJava有点大才小用，没有用到精髓的地方。现在重新了解去认识RxJava，学习里面的一些思维，可以以后在项目中、自己的代码中使用其思维。不是只会里面的形，而没有领会到其神。
基本的订阅流程
简单的先从其基本的订阅流程来追踪源码。
下图是自己梳理画的订阅流程图：
先来看看基本的订阅代码：
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("Hello");
emitter.onNext("World");
emitter.onNext("!");
emitter.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
Log.i("qhh",s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.v("qhh","complete");
}
});
Observable 被观察者
Observable中需要关注三个方法:
- 静态方法 create
Observable调用create方法创建第一个Observable，先看看create方法中的源码实现。
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
create方法的入参是 ObservableOnSubscribe ，这个又是什么呢？进一步简单的看一下源码。
public interface ObservableOnSubscribe<T> {
/**
* Called for each Observer that subscribes.
* @param emitter the safe emitter instance, never null
* @throws Exception on error
*/
void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}
是一个接口，仅有一个方法 subscribe 。这个就很有意思了，通过他的注释和传入的参数，猜测这里的发送事件的逻辑是直接和Observer对象有关系，虽然没有直接的出现Observer。
在源码中还发现一个类 ObservableCreate 。这个类是继承了 Observable 。也就是 create 创建了第一个 Observable 。并且在 ObservableCreate中持有 ObservableOnSubscribe 对象。还发现了 subscribeActual 方法。也就是最终的订阅实现在这里实现。
@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);
}
}
而该方法是复写了 Observable中的方法。那又是什么时候开始调用这个方法呢？带着疑惑继续往下看。跳出当前源码，紧接着看 subscribe 方法。
- subscribe
Observable中的subscribe方法入参则是一个Observer，进入源码看看他是如何处理这个传入的Observer。
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
.....
} catch (Throwable e) {
.....
}
}
最终看到其内部调用了 subscribeActual ，将 Observer对象传入到 ObservableCreate（Observable子类）中，则在 subscribeActual 进行业务实现。
- subscribeActual
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);
}
}
....
}
CreateEmitter 是 Observer 的一个封装类，因此尅一看到最初示例中的订阅关系。 在 ObservableOnSubscribe 的方法 subscribe 中，emitter 调用 onNext 、onComplete 等方法，实际就是调用了在 subscribe 的时候传入的 Observer 对应的 onNext、onComplete等方法。
至此，简单的一个订阅流程就梳理了一下，但是感觉其思想类似MVP的一些思路。ObservableCreate 相当于P这个桥梁，都持有 ObservableOnSubscribe 和 Observer 对象，进而进行相互的传递操作。
简单的示例图：
