最近有时间，研究了一下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经典代码的回调。
