RxJava
前言
Linus Benedict Torvalds : RTFSC – Read The Fucking Source Code
概括
原理解析我们从三个部分入手：1、基本流程梳理，2、lift的实现原理，3、线程切换原理。
基本流程
Observable.create(new Observable.OnSubscribe<ArrayList<String>>() {
@Override
public void call(Subscriber<? super ArrayList<String>> subscriber) {
subscriber.onNext(T);
subscriber.onCompleted();
}
})
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(String s) {}
});
这就是Rxjava的基本使用流程。我们从流程开始分析：
Observable.create()
// Step 1:
public final static <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(hook.onCreate(f));
}
// Step 2:
protected Observable(OnSubscribe<T> f) {
this.onSubscribe = f;
}
这两步是创建了一个Observable对象和OnSubscribe对象。
Observable.subscribe()
private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
// new Subscriber so onStart it
subscriber.onStart();
// allow the hook to intercept and/or decorate
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
}
这一步的操作实际上就是调用了 Observable.onSubscribe.call(subscriber);
整理下基本流程分两步：
1、创建Observable, OnSubscribe, Subscriber 三个对象
2、调用Observable的subscribe方法
那么他们最终的程序调用顺序是：
1、Observable.subscribe()
2、Observable.OnSubscribe.call()
3、Subscriber.onNext()
map操作符和lift原理
RxJava之所以好用是因为它提供了很多非常简洁的操作符动作，利用lift原理可以很流畅的实现操作的顺序过度。
Observable.create(new Observable.OnSubscribe<ArrayList<String>>() {
@Override
public void call(Subscriber<? super ArrayList<String>> subscriber) {
subscriber.onNext(T);
subscriber.onCompleted();
}
})
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return null;
}
})
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(String s) {}
});
Observable.map()
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
return lift(new OperatorMap<T, R>(func));
}
其实可以看到我们的map函数的主要实现是调用了litf方法，并在里面传入的是OperatorMap的对象。
Observable lift()
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return new Observable<R>(new OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> o) {
Subscriber<? super T> st = hook.onLift(operator).call(o);
st.onStart();
onSubscribe.call(st);
}
});
}
- 首先 new 了一个 Func函数, 保存在了
OperatorMap.transform中。 - new了一个新的 Observable. 这个 Observable 的构造函数中, 传入了一个新的 OnSubscribe。
lift的重点就在这个OnSubscribe里面。
先来看下OperatorMap的代码(只列出核心)：
public final class OperatorMap<T, R> implements Operator<R, T> {
@Override
public Subscriber<? super T> call(final Subscriber<? super R> o) {
return new Subscriber<T>(o) {
@Override
public void onNext(T t) {
o.onNext(transformer.call(t));
}
};
}
}
我们看到从hook.onLift(operator).call(o);这里创建出来的Subscriber的call函数里主要做的就是transformer.call(t)也就是转换的操作。然后再将其转换的结果返回给onNext。
lift 流程总结
总结下整个回调流程：
- 调用Create函数，创建一个新的Observable，并且创建一个OnSubscribe。
- 调用Map函数，再创建一个新的Observable，同时也创建一个新的OnSubscribe。并且其中的
OnSubscribe.call()方法调用了第一步的Observable.onSubscribe.call方法，最后将结果进行transform操作。 - 最后将拿到的
transform操作结果传给我们最后的onNext()方法中作为结果。
subscribeOn()和observeOn()线程切换原理
Rxjava还有另外一个原因让它在使用上非常简洁，就是线程处理上。
首先来看下他们的作用域
Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.map(mapOperator) // 新线程，由 observeOn() 指定
.observeOn(Schedulers.io())
.map(mapOperator2) // IO 线程，由 observeOn() 指定
.observeOn(AndroidSchedulers.mainThread)
.subscribe(subscriber); // Android 主线程，由 observeOn() 指定
我们先上一个例子：
Observable.create(new Observable.OnSubscribe<ArrayList<String>>() {
@Override
public void call(Subscriber<? super ArrayList<String>> subscriber) {
subscriber.onNext(T);
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(String s) {}
});
首先来看下observeOn()代码：
public final Observable<T> observeOn(Scheduler scheduler) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return lift(new OperatorObserveOn<T>(scheduler));
}
public Subscriber<? super T> call(Subscriber<? super T> child) {
if (scheduler instanceof ImmediateScheduler) {
return child;
} else if (scheduler instanceof TrampolineScheduler) {
return child;
} else {
ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child);
parent.init();
return parent;
}
}
可以看到observeOn其实也是调用了OperatorObserveOn类来实现的，本意上所有的Rxjava都是以操作符的形式来调用。接下来就是OperatorObserveOn的call()方法，该方法回根据传入的Subscriber类型决定返回的方式。
总结来说就是通过lift()方法将接下来传入执行的subscribe方法通过新的线程方式执行，并且在结束后再接着往下走。
总结
Rxjava 的原理我们并没有每条都往里深入，因为rxjava的代码复杂度很高，需要跟着代码有耐心的往下走才能看清，建议读者可以根据源码一步步跟进。
