先上代碼：
ObservableOnSubscribe<Integer> oos = new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
final int max = 100;
for (int i = 1; i <= max; i++) {
e.onNext(max);
}
e.onComplete();
}
};
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
Observable.create(oos)
.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.computation())
.subscribe(observer);
上面是Rxjava最簡(jiǎn)單的實(shí)現(xiàn)模型。
從鏈?zhǔn)秸{(diào)用的返回值來看：
Observable.create()------》ObservableCreate extends Observable
ObservableCreate.observerOn()------->ObservableObserveOn extends AbstractObservableWithUpstream extends Observable
ObservableObserveOn.subscribeOn()------->ObservableSubscribeOn extends AbstractObservableWithUpstream extends Observable
所以最后的調(diào)用對(duì)象是
ObservableSubscribeOn.subscribe(observer)
從上面的返回值可以看出中間任一一個(gè)的返回值返回的都是observable的子對(duì)象。
為什么要強(qiáng)調(diào)中間幾個(gè)的返回值都是observable的返回值，這里要先明確一下，待會(huì)會(huì)大量用到subscribe()方法，在Observable（子類）中的subscribe()方法：
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
//鉤子，如果未設(shè)置的話，返回值還是observer
observer = RxJavaPlugins.onSubscribe(this, observer);
//空檢驗(yàn)
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
//核心
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
...... //handle exception
}
}
因此下面的分析代碼中，如果是調(diào)用上面4個(gè)對(duì)象的subscribe()方法的時(shí)候，直接看subscribeActual()方法即可。
那就從最后一層 ObservableSubscribeOn 的 subscribeActual() 方法開始分析。
@Override
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
這里的這個(gè)s就是我們縮寫的observer
第一行：先把我們的observer封裝成了SubscribeOnObserver
第二行：調(diào)用了observer.onSubscribe()方法
??????????????也就是observer訂閱Observable時(shí)候的方法，一般這個(gè)時(shí)候可以做一些操作
第三行：
parent.setDisposable() 以及scheduler調(diào)度器先不論，待會(huì)再分析，這里先看SubscribeTask這個(gè)類：
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
source.subscribe(parent);
}
}
看到runnable,熟悉線程的同學(xué)已經(jīng)可以猜到source.subscribe(parent) 這句代碼很可能在子線程中執(zhí)行，這里先mark一下，待會(huì)回到這個(gè)地方再具體看。
這里要先插入一下source和Observer的問題：
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
//后面兩個(gè)類繼承抽象類，調(diào)用super(source)方法。
AbstractObservableWithUpstream(ObservableSource<T> source) {
this.source = source;
}
通過上面的代碼可以看到所有的這三個(gè)關(guān)鍵類，source都是通過構(gòu)造傳入進(jìn)來的，
而后兩個(gè)類都還有schedule參數(shù)，這個(gè)涉及線程調(diào)度，待會(huì)也會(huì)說，也mark一下。
通過以上代碼可以分析：
ObservableCreate 的 source 是 oos
ObservableObserveOn 的 source 是 ObservableCreate
ObservableSubscribeOn 的 source 是 ObservableObserveOn
至于Observer，通過代碼可以分析：
(ObservableObserveOn)source
.subscribe(parent(SubscribeOnObserver));
(ObservableCreate)source
.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
(oos)source.subscribe(parent);
也即是：
ObservableObserveOn 的 observer 是 SubscribeOnObserver
ObservableCreate 的 observer 是 ObserveOnObserver
oos 的 observer 是 CreateEmitter
這里有點(diǎn)一級(jí)一級(jí)調(diào)用的意味了，而這個(gè)意味就是Rxjava的一個(gè)很重要的點(diǎn)。
插入結(jié)束，繼續(xù)回到剛才的 SubscribeTask
結(jié)合上面的分析：
source.subscribe(parent)
也就意味著
ObservableObserveOn.subscribeActual()
這里轉(zhuǎn)了兩個(gè)彎，各位可以稍微思考一下
而在ObservableObserveOn中：
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
//這個(gè)暫時(shí)忽略，未設(shè)置的時(shí)候不走這里
source.subscribe(observer);
} else {
//線程調(diào)度，待會(huì)再分析
Scheduler.Worker w = scheduler.createWorker();
//最終會(huì)調(diào)用這個(gè)，又是很熟悉的source subscribe()方法
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
繼續(xù)往上走，走到 ObservableCreate 中，這里省略了重復(fù)流程。
@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);
}
}
走到最上層，這個(gè)最上層的source就是我們前面寫的oos。
這個(gè)observer是在 ObservableObserverOn 中的 ObserveOnObserver。這個(gè)名字有點(diǎn)像，汗
第一行：先把 ObserveOnObserver 封裝成 CreateEmitter
第二行：調(diào)用 ObserveOnObserver.onSubscribe()方法。
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
if (s instanceof QueueDisposable) {
@SuppressWarnings("unchecked")
QueueDisposable<T> qd = (QueueDisposable<T>) s;
int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
if (m == QueueDisposable.SYNC) {
sourceMode = m;
queue = qd;
done = true;
actual.onSubscribe(this);
schedule();
return;
}
if (m == QueueDisposable.ASYNC) {
sourceMode = m;
queue = qd;
actual.onSubscribe(this);
return;
}
}
queue = new SpscLinkedArrayQueue<T>(bufferSize);
actual.onSubscribe(this);
}
}
這個(gè)方法比較長(zhǎng)，但是對(duì)我們的流程分析關(guān)鍵的代碼其實(shí)就一句
actual.onSubscribe(this);
根據(jù)前面的observer的分析，這個(gè)observer其實(shí)就是 ObservableSubscribeOn 的 SubscribeOnObserver
最后找到源碼，調(diào)用了 SubscribeOnObserver 的 onSubscribe()方法。
@Override
public void onSubscribe(Disposable s) {
DisposableHelper.setOnce(this.s, s);
}
public static boolean setOnce(AtomicReference<Disposable> field, Disposable d) {
ObjectHelper.requireNonNull(d, "d is null");
if (!field.compareAndSet(null, d)) {
d.dispose();
if (field.get() != DISPOSED) {
reportDisposableSet();
}
return false;
}
return true;
}
涉及到CAS的操作，感興趣的同學(xué)可以研究一下，這里對(duì)我們的流程沒有太大影響。
第三行：至此，整個(gè)的流程終于回到了我們的oos。
從ObservableSubscribeOn的subscribe()方法歷盡千辛萬苦終于調(diào)用了oos的subscribe()方法。
ObservableOnSubscribe<Integer> oos = new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
final int max = 100;
for (int i = 1; i <= max; i++) {
e.onNext(max);
}
e.onComplete();
}
};
首先創(chuàng)建了 ObservableEmitter ，然后調(diào)用emmiter.onNext()方法。
@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()) {
observer.onNext(t);
}
}
異常先不考慮，最終是調(diào)用 observer.onNext()方法。
根據(jù)上面的分析，這個(gè)Observer是ObserveOnObserver
第一行：先把 ObserveOnObserver 封裝成 CreateEmitter，而CreateEmmiter的構(gòu)造：
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
所以，可以知道這個(gè)observer就是最前面的 ObserveOnObserver
也就是e.Next(n)------>最終會(huì)調(diào)用ObserveOnObserver.onNext(n)
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
最終調(diào)用了 schedule() 方法。
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
將當(dāng)前對(duì)象添加到worker中，這個(gè)是線程調(diào)度的問題了，待會(huì)分析。
再看一下ObserveOnObserver 類的聲明：
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable{}
實(shí)現(xiàn)了Runnable接口，所以關(guān)鍵代碼就在run()方法之中。
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
//我們看drainNormal()的方法
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
終于看到了a.onNext()方法，也就是actual.onNext()方法。
通過 ObserveOnObserver 的構(gòu)造：
//構(gòu)造方法
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.actual = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
//創(chuàng)建對(duì)象的時(shí)候
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
//這里new了ObserveOnObserver對(duì)象。
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
可以看出，這個(gè)actual對(duì)象，其實(shí)就是傳入進(jìn)來的observer。
而這個(gè)observer結(jié)合SubscribeTask代碼，可以知道：
這個(gè)observer其實(shí)就是講我們的observer封裝起來的SubscribeOnObserver對(duì)象。
而SubscribeOnObserver的onNext()方法：
@Override
public void onNext(T t) {
actual.onNext(t);
}
其實(shí)就是我們的o.next()方法。
七轉(zhuǎn)八彎，經(jīng)歷這個(gè)這么多,也是本文最核心的：
subscribe()方法，先一層一層往上回調(diào)，調(diào)用了我們的oos的onNext()方法，
而onNext()里面又一層一層往下回調(diào)，調(diào)用了我們的obsrever的onNext()方法，實(shí)現(xiàn)了數(shù)據(jù)的傳遞。
然后是線程切換問題：
還記得我們之前說ObservableSubscribeOn, ObservableObserveOn這兩個(gè)對(duì)象的構(gòu)造都會(huì)傳入一個(gè) schedule 的調(diào)度器嗎？
先看 ObservableSubscribeOn
@Override
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
再結(jié)合前面的代碼，我們知道這個(gè) scheduler 是Schedulers.computation()
最后走到了：
public Disposable scheduleDirect(final Runnable run, long delayTime, TimeUnit unit) {
ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run));
try {
Future<?> f;
if (delayTime <= 0L) {
f = executor.submit(task);
} else {
f = executor.schedule(task, delayTime, unit);
}
task.setFuture(f);
return task;
} catch (RejectedExecutionException ex) {
RxJavaPlugins.onError(ex);
return EmptyDisposable.INSTANCE;
}
}
executor,我們非常熟悉的線程池。看到這，也就大概明白了我們的 source.subscribe(parent)
以及其對(duì)應(yīng)的一層層往上回調(diào)都是在subscribeOn(線程) 所調(diào)用的線程之中。
然后線程什么時(shí)候會(huì)再度切換呢？
是在ObservableObserveOn中的 schedule() 方法中：
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
這個(gè)worker一層層追蹤溯源，找到了其初始化的地方，是在ObservableObserveOn的subscribeActual()方法之中：
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
這個(gè)schedule就是observerOn所對(duì)應(yīng)的線程。
AndroidSchedulers.mainThread() 的實(shí)現(xiàn)是 HandlerScheduler
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException("run == null");
if (unit == null) throw new NullPointerException("unit == null");
if (disposed) {
return Disposables.disposed();
}
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
if (async) {
message.setAsynchronous(true);
}
handler.sendMessageDelayed(message, unit.toMillis(delay));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
最終通過hanler進(jìn)行了線程的切換。
也就是最后我們的observer.onNext()方法執(zhí)行的線程是由observeOn()所對(duì)應(yīng)的線程
