沒有前言，沒有介紹，直接開擼！！
RxJava在Android中使用，最常用的流程：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "--create subscribe: " + Thread.currentThread().getName());
Log.d(TAG, "--emitter subscribe: start");
emitter.onNext(1);
Log.d(TAG, "--emitter subscribe: over");
}
})
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
Log.d(TAG, "--map: " + Thread.currentThread().getName());
Log.d(TAG, "--map: receive = " + integer);
return "map " + integer;
}
})
.flatMap(new Function<String, ObservableSource<Boolean>>() {
@Override
public ObservableSource<Boolean> apply(String s) throws Exception {
Log.d(TAG, "--flatMap receive = " + s);
return Observable.just(true);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Boolean>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: " + Thread.currentThread().getName());
Log.d(TAG, "onSubscribe: ");
}
@Override
public void onNext(Boolean result) {
Log.d(TAG, "onNext: " + Thread.currentThread().getName());
Log.d(TAG, "onNext: ");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: ");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});
執(zhí)行的流程日志打印.png
1.Observable.create（ObservableOnSubscribe<Integer> source）
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//檢測參數(shù)是否為null，為null則拋出空指針異常
ObjectHelper.requireNonNull(source, "source is null");
//onAssembly（）是一個(gè)鉤子函數(shù)，這里不用管，實(shí)質(zhì)上return的就是ObservableCreate對象
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
創(chuàng)建一個(gè)ObservableOnSubscribe對象，實(shí)現(xiàn)subscribe(ObservableEmitter<Integer> emitter)方法，并將ObservableOnSubscribe對象，傳到ObservableCreate的構(gòu)造函數(shù)中去，并返回ObservableCreate對象，create()在這里執(zhí)行完畢；
來看ObservableCreate的構(gòu)造函數(shù)
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
//注意這個(gè)Source對象，后面的分析都會有類似的source
this.source = source;
}
}
2.Observable.map（Function<? super T, ? extends R> mapper）
//T表示上游onNext傳遞下來的數(shù)據(jù)類型，示例中是Integer，R表示轉(zhuǎn)換的數(shù)據(jù)類型，示例中是String
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
//鉤子函數(shù)，在這里本質(zhì)是返回ObservableMap對象
//注意：這個(gè)this，是指上面的ObservableCreate，mapper是我們創(chuàng)建的Function對象
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
來看下ObservableMap的構(gòu)造函數(shù)
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
//調(diào)用父類AbstractObservableWithUpstream的構(gòu)造函數(shù)，賦值source
super(source);
this.function = function;
//現(xiàn)在ObservableMap持有兩個(gè)對象，一個(gè)是ObservableCreate，一個(gè)是Function
}
}
3.Observable.flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper)
public final <R> Observable<R> flatMap(
Function<? super T, ? extends ObservableSource<? extends R>> mapper,
boolean delayErrors, int maxConcurrency, int bufferSize) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
//當(dāng)前的this，是指ObservableMap對象，這里跳過
if (this instanceof ScalarCallable) {
@SuppressWarnings("unchecked")
T v = ((ScalarCallable<T>)this).call();
if (v == null) {
return empty();
}
return ObservableScalarXMap.scalarXMap(v, mapper);
}
//同上，這里返回一個(gè)ObservableFlatMap對象
//參數(shù)：
//this：ObservableMap
//mapper：Function<String,ObservableSource<Boolean>>
//delayErroes:fasle
//maxConcurrency:Integer.MAX_VALUE
//bufferSize:bufferSize()
return RxJavaPlugins.onAssembly(new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize));
}
來看ObservableFlatMap構(gòu)造函數(shù)，同樣持有上游的ObservableMap對象，和我們創(chuàng)建的Function對象
public final class ObservableFlatMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends ObservableSource<? extends U>> mapper;
final boolean delayErrors;
final int maxConcurrency;
final int bufferSize;
public ObservableFlatMap(ObservableSource<T> source,
Function<? super T, ? extends ObservableSource<? extends U>> mapper,
boolean delayErrors, int maxConcurrency, int bufferSize) {
super(source);
this.mapper = mapper;
this.delayErrors = delayErrors;
this.maxConcurrency = maxConcurrency;
this.bufferSize = bufferSize;
}
}
4.Observable.subscribeOn(Scheduler scheduler)
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
//返回ObservableSubscribeOn對象
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
老規(guī)矩，看ObservableSubscribeOn構(gòu)造函數(shù)
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
//同樣持有上游ObservableFlatMap對象，和一個(gè)傳進(jìn)來的Scheduler對象
super(source);
this.scheduler = scheduler;
}
}
5.Observable. observeOn(Scheduler scheduler)
public final Observable<T> observeOn(Scheduler scheduler,
boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
//參數(shù)：
//this:ObservableSubscribeOn
//scheduler:AndroidSchedulers.mainThread()
//delayError:false
//bufferSize:bufferSize()
return RxJavaPlugins.onAssembly(
new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
繼續(xù)老規(guī)矩（感覺自己好啰嗦，為了把流程寫清楚，方便回顧，shit 0.0），看ObservableSubscribeOn構(gòu)造函數(shù)
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source,
Scheduler scheduler, boolean delayError, int bufferSize) {
//持有上游的ObservableSubscribeOn對象，以及我們傳入的AndroidSchedulers.mainThread()
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
}
到這里就只剩最后一個(gè)subscribe()方法，就此打?。?/p>
我們來總結(jié)一下，上面都發(fā)生了什么：
| Observable | 持有ObservableSource對象 | 持有其他對象 |
|---|---|---|
| ObservableCreate | ObservableOnSubscribe | |
| ObservableMap | ObservableCreate | Function |
| ObservableFlatMap | ObservableMap | Function |
| ObservableSubscribeOn | ObservableFlatMap | Schedulers.io() |
| ObservableObserveOn | ObservableSubscribeOn | AndroidSchedulers.mainThread() |
有沒有發(fā)現(xiàn)，這五個(gè)對象都繼承Observable，都有subscribe()方法，實(shí)現(xiàn)了Observable的subscribeActual抽象方法，當(dāng)調(diào)用subscribe()時(shí)，則會一層層的進(jìn)行回傳調(diào)用，接下來，咱們繼續(xù)分析調(diào)用subscribe()后的流程
6.Observable. subscribe(Observer<? super T> observer)
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
//這個(gè)也是個(gè)鉤子函數(shù)，暫且不管
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "The RxJavaPlugins...");
//重點(diǎn)在這里，subscribeActual是Observable的抽象函數(shù)
//回到最后的Observable對象：ObservableObserveOn
subscribeActual(observer);
} catch (NullPointerException e) {
...
}
}
7.回到ObservableObserveOn.subscribeActual()
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
@Override
protected void subscribeActual(Observer<? super T> observer) {
//這里我們傳入的scheduler是AndroidSchedulers.mainThread()
//本質(zhì)是HandlerScheduler，所以進(jìn)入else分支
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
//source是指ObservableSubscribeOn，將我們創(chuàng)建的observer進(jìn)行包裝
//包裝成ObserveOnObserver對象，傳給ObservableSubscribeOn
//回到ObservableSubscribeOn
source.subscribe(
new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
ObserveOnObserver(Observer<? super T> actual,
Scheduler.Worker worker, boolean delayError, int bufferSize) {
//這里要注意，downstream是指下游的傳進(jìn)來的Observer對象
this.downstream = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
}
}
8.回到ObservableSubscribeOn.subscribeActual()
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
@Override
public void subscribeActual(final Observer<? super T> observer) {
//將下游的observer進(jìn)行包裝，生成SubscribeOnObserver對象
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
//下游的ObserveOnObserver調(diào)用onSubscribe方法，回到8.1
observer.onSubscribe(parent);
//8.2查看SubscribeTask類
//8.3查看scheduler.scheduleDirect()方法
//8.4查看ObservableSubscribeOn.setDisposable()方法
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
SubscribeOnObserver(Observer<? super T> downstream) {
//這里的downstream，是指ObserveOnObserver對象
this.downstream = downstream;
this.upstream = new AtomicReference<Disposable>();
}
}
}
8.1ObservableObserveOn的內(nèi)部類ObserveOnObserver的onSubscribe方法
public void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
//將上游的SubscribeOnObserver進(jìn)行保存
this.upstream = d;
//SubscribeOnObserver未實(shí)現(xiàn)QueueDisposable接口，這里跳過
if (d instanceof QueueDisposable) {
...
}
queue = new SpscLinkedArrayQueue<T>(bufferSize);
//調(diào)用下游的Observe對象，及我們創(chuàng)建的Observe，這時(shí)打印我們的第一行日志
downstream.onSubscribe(this);
}
}
//DisposableHelper.validate（）
public static boolean validate(Disposable current, Disposable next) {
//此時(shí)的current為null，next不為null，返回true
if (next == null) {
RxJavaPlugins.onError(new NullPointerException("next is null"));
return false;
}
if (current != null) {
next.dispose();
reportDisposableSet();
return false;
}
return true;
}
8.2查看SubscribeTask類
//SubscribeTask實(shí)現(xiàn)了Runnable
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
//在這里繼續(xù)向上訂閱，調(diào)用ObservableFlatMap的subscribe方法
//多說一句，到這里關(guān)系到RxJava的線程切換，這里是將一個(gè)Runnable對象傳入Scheduler
//如果我們外部傳入的Scheduler是Schedulers.io(),那么這里在子線程執(zhí)行
//如果我們外部傳入的Scheduler是AndroidSchedulers.mainThread(),那么這里在UI線程執(zhí)行
//這里也可以知道，為什么我們調(diào)用多次.subscribeOn()，只有后最上層的那個(gè)起作用
source.subscribe(parent);
}
}
8.3查看scheduler.scheduleDirect()方法
public Disposable scheduleDirect(
@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
DisposeTask task = new DisposeTask(decoratedRun, w);
//主要看這兒
w.schedule(task, delay, unit);
//返回Disposable對象
return task;
}
//IoScheduler內(nèi)部類EventLoopWorker的schedule方法
public Disposable schedule(
@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
return EmptyDisposable.INSTANCE;
}
//看這兒，這才是真正執(zhí)行的scheduleActual
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
//NewThreadWorker
public ScheduledRunnable scheduleActual(final Runnable run,
long delayTime,
@NonNull TimeUnit unit,
@Nullable DisposableContainer parent) {
//這個(gè)run，就是我們最開始傳進(jìn)來的那個(gè)
//鉤子函數(shù)
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}
//這里直接將run放入到線程池執(zhí)行
Future<?> f;
try {
if (delayTime <= 0) {
f = executor.submit((Callable<Object>)sr);
} else {
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {
if (parent != null) {
parent.remove(sr);
}
RxJavaPlugins.onError(ex);
}
return sr;
}
8.4 SubscribeOnObserver.setDisposable()方法
void setDisposable(Disposable d) {
//這里有點(diǎn)沒搞明白，看了“原件”的簡書關(guān)于RxJava源碼分析，是這樣說的
//上游的onSubscribe會調(diào)用，但是因?yàn)閠his.s的disposable不為null，大部分情況一直都是直接跳過
DisposableHelper.setOnce(this, d);
}
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;
}
9.ObservableFlatMap.subscribe()
public void subscribeActual(Observer<? super U> t) {
if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
return;
}
//source是ObservableMap對象，
//t是SubscribeOnObserver對象，
//將t進(jìn)行包裝，并將我們創(chuàng)建的Function一并 傳入MergeObserver
source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}
啰嗦系列+1，看MergeObserver構(gòu)造函數(shù)
//ObservableFlatMap靜態(tài)內(nèi)部類
static final class MergeObserver<T, U> extends AtomicInteger implements Disposable, Observer<T> {
MergeObserver(Observer<? super U> actual,
Function<? super T, ? extends ObservableSource<? extends U>> mapper,
boolean delayErrors, int maxConcurrency, int bufferSize) {
//下游的Observer，這里是SubscribeOnObserver
this.downstream = actual;
//這里是我們創(chuàng)建的Function
this.mapper = mapper;
this.delayErrors = delayErrors;
this.maxConcurrency = maxConcurrency;
this.bufferSize = bufferSize;
if (maxConcurrency != Integer.MAX_VALUE) {
sources = new ArrayDeque<ObservableSource<? extends U>>(maxConcurrency);
}
this.observers = new AtomicReference<InnerObserver<?, ?>[]>(EMPTY);
}
}
10.ObservableMap.subscribe()
public void subscribeActual(Observer<? super U> t) {
//這里同上，調(diào)用上游ObservableCreate的subscribe
//包裝下游的MergeObserver，生成當(dāng)前的MapObserver，傳給上游Observable
source.subscribe(new MapObserver<T, U>(t, function));
}
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
//注意這里，mapper就是我們想要轉(zhuǎn)化類型執(zhí)行的類，等下會用到
this.mapper = mapper;
}
}
11.ObservableCreate.subscribe()
protected void subscribeActual(Observer<? super T> observer) {
//observer是下游傳來的MapObserver
//CreateEmitter是ObservableCreate的內(nèi)部類，實(shí)現(xiàn)了ObservableEmitter，Disposable接口
//發(fā)射器調(diào)用onNext(1)發(fā)射時(shí)，會調(diào)observer的onNext
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//下游的observer訂閱發(fā)射器
observer.onSubscribe(parent);
try {
//調(diào)用我們創(chuàng)建的ObservableOnSubscribe對象的subscribe方法，通過發(fā)射器調(diào)用onNext
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
再次打住，到這里，通過底層ObservableObserveOn調(diào)用subscribe()方法，一層層的向上傳遞到頂層的ObservableSource對象，下一步就是發(fā)射對象，來擼一擼順序：
| ObservableSource | 持有的ObservableSource | 向上傳遞的Observer |
|---|---|---|
| ObservableObserveOn | ObservableSubscribeOn | ObserveOnObserver |
| ObservableSubscribeOn | ObservableFlatMap | SubscribeOnObserver |
| ObservableFlatMap | ObservableMap | MergeObserver |
| ObservableMap | ObservableCreate | MapObserver |
| ObservableCreate |
到頂層ObservableCreate，持有一個(gè)我們創(chuàng)建的ObservableOnSubscribe對象，持有下游的MapObserver對象，通過ObservableCreate的靜態(tài)內(nèi)部類CreateEmitter發(fā)射器，將數(shù)據(jù)發(fā)射源和下游關(guān)聯(lián)起來，我們接著往下看；
12.MapObserver.onSubscribe()
//其實(shí)是執(zhí)行的父類的onSubscribe
public final void onSubscribe(Disposable d) {
//進(jìn)行校驗(yàn)，這里返回為true
if (DisposableHelper.validate(this.upstream, d)) {
//將上游的Disposable對象保存，這里是指CreateEmitter發(fā)射器
this.upstream = d;
if (d instanceof QueueDisposable) {
this.qd = (QueueDisposable<T>)d;
}
//beforeDownstream()默認(rèn)返回true
if (beforeDownstream()) {
//這里的downstream，是指下游的MergeObserver對象,接著往下看
downstream.onSubscribe(this);
//afterDownstream()方法體為空
afterDownstream();
}
}
}
13.MergeObserver.onSubscribe()
public void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
this.upstream = d;
//繼續(xù)調(diào)用下游Observer，這里調(diào)用的是SubscribeOnObserver
downstream.onSubscribe(this);
}
}
14.SubscribeOnObserver.onSubscribe()
public void onSubscribe(Disposable d) {
//又調(diào)了這個(gè)方法，回到ObservableCreate繼續(xù)往下看
DisposableHelper.setOnce(this.upstream, d);
}
15.開始執(zhí)行發(fā)射
層層調(diào)用onSubscribe之后，開始進(jìn)入發(fā)射
try {
//這個(gè)source，是我們創(chuàng)建的ObservableOnSubscribe對象，將CreateEmitter發(fā)射器傳進(jìn)去
//并調(diào)用CreateEmitter.onNext(1)，發(fā)射一個(gè)數(shù)值1
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "--create subscribe: " + Thread.currentThread().getName());
Log.d(TAG, "--emitter subscribe: start");
emitter.onNext(1);
Log.d(TAG, "--emitter subscribe: over");
}
}
16.CreateEmitter.onNext()
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null..."));
return;
}
if (!isDisposed()) {
//發(fā)射器持有MapObserver對象，接著往下看
observer.onNext(t);
}
}
17.MapObserver.onNext()
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
//這里的mapper是我們創(chuàng)建的Function對象，執(zhí)行apply()方法，要求返回值不能為null
//U v，這是我們目標(biāo)類型
//這里就可以知道 map操作符 是怎么樣轉(zhuǎn)化數(shù)據(jù)類型的
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper...");
} catch (Throwable ex) {
fail(ex);
return;
}
//執(zhí)行成功后，接著往下游傳遞 類型轉(zhuǎn)換后的數(shù)據(jù)
//這里的downstream是指MergeObserver，接著來
downstream.onNext(v);
}
18.MergeObserver.onNext()
public void onNext(T t) {
// safeguard against misbehaving sources
if (done) {
return;
}
ObservableSource<? extends U> p;
try {
//這里的mapper也是我們創(chuàng)建的Function對象，執(zhí)行apply()方法，要求返回值不能為null
//返回一個(gè)ObservableSource<? extends U>對象
//在示例中，我們返回的是：Observable.just(true) --->>
//實(shí)質(zhì)上是一個(gè)ObservableJust對象
p = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper...");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
upstream.dispose();
onError(e);
return;
}
//maxConcurrency值為Integer.MAX_VALUE，這里跳過
if (maxConcurrency != Integer.MAX_VALUE) {
synchronized (this) {
if (wip == maxConcurrency) {
sources.offer(p);
return;
}
wip++;
}
}
//接著往下看subscribeInner
subscribeInner(p);
}
void subscribeInner(ObservableSource<? extends U> p) {
for (;;) {
//我們返回的ObservableJust實(shí)現(xiàn)了ScalarCallable，Callable接口
if (p instanceof Callable) {
//我們重點(diǎn)關(guān)注tryEmitScalar這個(gè)方法
if (tryEmitScalar(((Callable<? extends U>)p))
&& maxConcurrency != Integer.MAX_VALUE) {
...
} else {
break;
}
}
...
}
}
boolean tryEmitScalar(Callable<? extends U> value) {
U u;
try {
//value還是ObservableJust，執(zhí)行call()方法，返回我們外部傳進(jìn)去的值：true
u = value.call();
} catch (Throwable ex) {
...
}
if (u == null) {
return true;
}
//MergeObserver 繼承了 AtomicInteger，利用同步機(jī)制，同時(shí)只會有一個(gè) value 被發(fā)射
if (get() == 0 && compareAndSet(0, 1)) {
//調(diào)用下游的Observer，這里是SubscribeOnObserver，接著往下看
downstream.onNext(u);
if (decrementAndGet() == 0) {
return true;
}
} else {
...
}
drainLoop();
return true;
}
19.SubscribeOnObserver.onNext()
public void onNext(T t) {
//這就很簡單了，繼續(xù)往下傳遞，這個(gè)downstream是指ObserveOnObserver
downstream.onNext(t);
}
20.ObserveOnObserver.onNext()
public void onNext(T t) {
if (done) {
return;
}
//這里的sourceMode并未被賦值，跳過
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
//重點(diǎn)看這個(gè)
schedule();
}
void schedule() {
if (getAndIncrement() == 0) {
//這個(gè)worker是我們傳進(jìn)來的AndroidSchedulers.mainThread()生成的Worker對象
//實(shí)質(zhì)是HandlerWorker對象，接著看
worker.schedule(this);
}
}
21.HandlerWorker.schedule()
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);
//將傳進(jìn)來的run進(jìn)行包裝，run本質(zhì)是ObserveOnObserver
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
//這個(gè)看著來勁吧，RxAndroid在Android中的線程切換，最終還是通過handler來進(jìn)行的，真相大白
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);
}
//這里發(fā)送一個(gè)message，所以run里的執(zhí)行，都是在UI線程
//回到ObserveOnObserver，看run()方法里有哪些東西
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;
}
22.ObserveOnObserver.run()
public void run() {
if (outputFused) {
drainFused();
} else {
//本示例執(zhí)行的這個(gè)方法
drainNormal();
}
}
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
//downstream是我們創(chuàng)建的Observer對象
final Observer<? super T> a = downstream;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
...
//到這里調(diào)用onNext(),打印最后兩句日志，over?。。。? a.onNext(v);
}
...
}
}
RxJava執(zhí)行流程.png
? 先別急，回顧一下，總結(jié)這個(gè)流程，我們在使用操作符的時(shí)候，每使用一次，都會創(chuàng)建操作符對應(yīng)的Observable對象，并將上一個(gè)Observable作為參數(shù)，傳進(jìn)構(gòu)造函數(shù)中，保存為source變量（除了最開始的create傳入的ObservableOnSubscribe不是Observable的子類）；
? 鏈?zhǔn)秸{(diào)用完操作符之后，最后的這個(gè)Observable調(diào)用subscribe訂閱時(shí)，調(diào)用subscribeActual抽象方法，首先創(chuàng)建該當(dāng)前Observable對象的Observer內(nèi)部類對象，進(jìn)行包裝，然后調(diào)用上游Source的subscribe方法，將Observer對象上傳給上游，直到傳到頂層；
? 頂層ObservableCreate，沒有Observer內(nèi)部類，而是發(fā)射器CreateEmitter，同樣有onNext()方法，在ObservableCreate的subscribeActual中，先調(diào)用onSubscribe，一層層往下調(diào)用，保存上游的Disposable，然后再調(diào)用我們創(chuàng)建的ObservableOnSubscribe.subscribe()，執(zhí)行發(fā)射器的onNext()，再一層層的往下調(diào)Observer，直到調(diào)用到我們創(chuàng)建的Observer對象，到此onNext的流程執(zhí)行完畢
媽耶，第一次寫這么多字，按照示例分析完onNext()之后，對RxJava的執(zhí)行流程有了小小的認(rèn)識，收獲很是挺多的，弄明白為啥subscribeOn只調(diào)用多次無用，子線程和主線程是怎樣切換的，操作符是怎樣轉(zhuǎn)化數(shù)據(jù)的，目前水平有限，需要提高的地方很多，大佬們多多指教，抱拳??！
