没有前言，没有介绍，直接开撸！！
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: ");
}
});
执行的流程日志打印.png
1.Observable.create（ObservableOnSubscribe<Integer> source）
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//检测参数是否为null，为null则抛出空指针异常
ObjectHelper.requireNonNull(source, "source is null");
//onAssembly（）是一个钩子函数，这里不用管，实质上return的就是ObservableCreate对象
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
创建一个ObservableOnSubscribe对象，实现subscribe(ObservableEmitter<Integer> emitter)方法，并将ObservableOnSubscribe对象，传到ObservableCreate的构造函数中去，并返回ObservableCreate对象，create()在这里执行完毕；
来看ObservableCreate的构造函数
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
//注意这个Source对象，后面的分析都会有类似的source
this.source = source;
}
}
2.Observable.map（Function<? super T, ? extends R> mapper）
//T表示上游onNext传递下来的数据类型，示例中是Integer，R表示转换的数据类型，示例中是String
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
//钩子函数，在这里本质是返回ObservableMap对象
//注意：这个this，是指上面的ObservableCreate，mapper是我们创建的Function对象
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
来看下ObservableMap的构造函数
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) {
//调用父类AbstractObservableWithUpstream的构造函数，赋值source
super(source);
this.function = function;
//现在ObservableMap持有两个对象，一个是ObservableCreate，一个是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");
//当前的this，是指ObservableMap对象，这里跳过
if (this instanceof ScalarCallable) {
@SuppressWarnings("unchecked")
T v = ((ScalarCallable<T>)this).call();
if (v == null) {
return empty();
}
return ObservableScalarXMap.scalarXMap(v, mapper);
}
//同上，这里返回一个ObservableFlatMap对象
//参数：
//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构造函数，同样持有上游的ObservableMap对象，和我们创建的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));
}
老规矩，看ObservableSubscribeOn构造函数
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
//同样持有上游ObservableFlatMap对象，和一个传进来的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");
//参数：
//this:ObservableSubscribeOn
//scheduler:AndroidSchedulers.mainThread()
//delayError:false
//bufferSize:bufferSize()
return RxJavaPlugins.onAssembly(
new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
继续老规矩（感觉自己好啰嗦，为了把流程写清楚，方便回顾，shit 0.0），看ObservableSubscribeOn构造函数
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;
}
}
到这里就只剩最后一个subscribe()方法，就此打住；
我们来总结一下，上面都发生了什么：
| Observable | 持有ObservableSource对象 | 持有其他对象 |
|---|---|---|
| ObservableCreate | ObservableOnSubscribe | |
| ObservableMap | ObservableCreate | Function |
| ObservableFlatMap | ObservableMap | Function |
| ObservableSubscribeOn | ObservableFlatMap | Schedulers.io() |
| ObservableObserveOn | ObservableSubscribeOn | AndroidSchedulers.mainThread() |
有没有发现，这五个对象都继承Observable，都有subscribe()方法，实现了Observable的subscribeActual抽象方法，当调用subscribe()时，则会一层层的进行回传调用，接下来，咱们继续分析调用subscribe()后的流程
6.Observable. subscribe(Observer<? super T> 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...");
//重点在这里，subscribeActual是Observable的抽象函数
//回到最后的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()
//本质是HandlerScheduler，所以进入else分支
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
//source是指ObservableSubscribeOn，将我们创建的observer进行包装
//包装成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是指下游的传进来的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进行包装，生成SubscribeOnObserver对象
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
//下游的ObserveOnObserver调用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的内部类ObserveOnObserver的onSubscribe方法
public void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
//将上游的SubscribeOnObserver进行保存
this.upstream = d;
//SubscribeOnObserver未实现QueueDisposable接口，这里跳过
if (d instanceof QueueDisposable) {
...
}
queue = new SpscLinkedArrayQueue<T>(bufferSize);
//调用下游的Observe对象，及我们创建的Observe，这时打印我们的第一行日志
downstream.onSubscribe(this);
}
}
//DisposableHelper.validate（）
public static boolean validate(Disposable current, Disposable next) {
//此时的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实现了Runnable
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
//在这里继续向上订阅，调用ObservableFlatMap的subscribe方法
//多说一句，到这里关系到RxJava的线程切换，这里是将一个Runnable对象传入Scheduler
//如果我们外部传入的Scheduler是Schedulers.io(),那么这里在子线程执行
//如果我们外部传入的Scheduler是AndroidSchedulers.mainThread(),那么这里在UI线程执行
//这里也可以知道，为什么我们调用多次.subscribeOn()，只有后最上层的那个起作用
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内部类EventLoopWorker的schedule方法
public Disposable schedule(
@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
return EmptyDisposable.INSTANCE;
}
//看这儿，这才是真正执行的scheduleActual
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
//NewThreadWorker
public ScheduledRunnable scheduleActual(final Runnable run,
long delayTime,
@NonNull TimeUnit unit,
@Nullable DisposableContainer parent) {
//这个run，就是我们最开始传进来的那个
//钩子函数
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}
//这里直接将run放入到线程池执行
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) {
//这里有点没搞明白，看了“原件”的简书关于RxJava源码分析，是这样说的
//上游的onSubscribe会调用，但是因为this.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进行包装，并将我们创建的Function一并 传入MergeObserver
source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}
啰嗦系列+1，看MergeObserver构造函数
//ObservableFlatMap静态内部类
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;
//这里是我们创建的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) {
//这里同上，调用上游ObservableCreate的subscribe
//包装下游的MergeObserver，生成当前的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就是我们想要转化类型执行的类，等下会用到
this.mapper = mapper;
}
}
11.ObservableCreate.subscribe()
protected void subscribeActual(Observer<? super T> observer) {
//observer是下游传来的MapObserver
//CreateEmitter是ObservableCreate的内部类，实现了ObservableEmitter，Disposable接口
//发射器调用onNext(1)发射时，会调observer的onNext
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//下游的observer订阅发射器
observer.onSubscribe(parent);
try {
//调用我们创建的ObservableOnSubscribe对象的subscribe方法，通过发射器调用onNext
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
再次打住，到这里，通过底层ObservableObserveOn调用subscribe()方法，一层层的向上传递到顶层的ObservableSource对象，下一步就是发射对象，来撸一撸顺序：
| ObservableSource | 持有的ObservableSource | 向上传递的Observer |
|---|---|---|
| ObservableObserveOn | ObservableSubscribeOn | ObserveOnObserver |
| ObservableSubscribeOn | ObservableFlatMap | SubscribeOnObserver |
| ObservableFlatMap | ObservableMap | MergeObserver |
| ObservableMap | ObservableCreate | MapObserver |
| ObservableCreate |
到顶层ObservableCreate，持有一个我们创建的ObservableOnSubscribe对象，持有下游的MapObserver对象，通过ObservableCreate的静态内部类CreateEmitter发射器，将数据发射源和下游关联起来，我们接着往下看；
12.MapObserver.onSubscribe()
//其实是执行的父类的onSubscribe
public final void onSubscribe(Disposable d) {
//进行校验，这里返回为true
if (DisposableHelper.validate(this.upstream, d)) {
//将上游的Disposable对象保存，这里是指CreateEmitter发射器
this.upstream = d;
if (d instanceof QueueDisposable) {
this.qd = (QueueDisposable<T>)d;
}
//beforeDownstream()默认返回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;
//继续调用下游Observer，这里调用的是SubscribeOnObserver
downstream.onSubscribe(this);
}
}
14.SubscribeOnObserver.onSubscribe()
public void onSubscribe(Disposable d) {
//又调了这个方法，回到ObservableCreate继续往下看
DisposableHelper.setOnce(this.upstream, d);
}
15.开始执行发射
层层调用onSubscribe之后，开始进入发射
try {
//这个source，是我们创建的ObservableOnSubscribe对象，将CreateEmitter发射器传进去
//并调用CreateEmitter.onNext(1)，发射一个数值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()) {
//发射器持有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是我们创建的Function对象，执行apply()方法，要求返回值不能为null
//U v，这是我们目标类型
//这里就可以知道 map操作符 是怎么样转化数据类型的
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper...");
} catch (Throwable ex) {
fail(ex);
return;
}
//执行成功后，接着往下游传递 类型转换后的数据
//这里的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也是我们创建的Function对象，执行apply()方法，要求返回值不能为null
//返回一个ObservableSource<? extends U>对象
//在示例中，我们返回的是：Observable.just(true) --->>
//实质上是一个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实现了ScalarCallable，Callable接口
if (p instanceof Callable) {
//我们重点关注tryEmitScalar这个方法
if (tryEmitScalar(((Callable<? extends U>)p))
&& maxConcurrency != Integer.MAX_VALUE) {
...
} else {
break;
}
}
...
}
}
boolean tryEmitScalar(Callable<? extends U> value) {
U u;
try {
//value还是ObservableJust，执行call()方法，返回我们外部传进去的值：true
u = value.call();
} catch (Throwable ex) {
...
}
if (u == null) {
return true;
}
//MergeObserver 继承了 AtomicInteger，利用同步机制，同时只会有一个 value 被发射
if (get() == 0 && compareAndSet(0, 1)) {
//调用下游的Observer，这里是SubscribeOnObserver，接着往下看
downstream.onNext(u);
if (decrementAndGet() == 0) {
return true;
}
} else {
...
}
drainLoop();
return true;
}
19.SubscribeOnObserver.onNext()
public void onNext(T t) {
//这就很简单了，继续往下传递，这个downstream是指ObserveOnObserver
downstream.onNext(t);
}
20.ObserveOnObserver.onNext()
public void onNext(T t) {
if (done) {
return;
}
//这里的sourceMode并未被赋值，跳过
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
//重点看这个
schedule();
}
void schedule() {
if (getAndIncrement() == 0) {
//这个worker是我们传进来的AndroidSchedulers.mainThread()生成的Worker对象
//实质是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);
//将传进来的run进行包装，run本质是ObserveOnObserver
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
//这个看着来劲吧，RxAndroid在Android中的线程切换，最终还是通过handler来进行的，真相大白
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);
}
//这里发送一个message，所以run里的执行，都是在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 {
//本示例执行的这个方法
drainNormal();
}
}
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
//downstream是我们创建的Observer对象
final Observer<? super T> a = downstream;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
...
//到这里调用onNext(),打印最后两句日志，over！！！！
a.onNext(v);
}
...
}
}
RxJava执行流程.png
先别急，回顾一下，总结这个流程，我们在使用操作符的时候，每使用一次，都会创建操作符对应的Observable对象，并将上一个Observable作为参数，传进构造函数中，保存为source变量（除了最开始的create传入的ObservableOnSubscribe不是Observable的子类）；
链式调用完操作符之后，最后的这个Observable调用subscribe订阅时，调用subscribeActual抽象方法，首先创建该当前Observable对象的Observer内部类对象，进行包装，然后调用上游Source的subscribe方法，将Observer对象上传给上游，直到传到顶层；
顶层ObservableCreate，没有Observer内部类，而是发射器CreateEmitter，同样有onNext()方法，在ObservableCreate的subscribeActual中，先调用onSubscribe，一层层往下调用，保存上游的Disposable，然后再调用我们创建的ObservableOnSubscribe.subscribe()，执行发射器的onNext()，再一层层的往下调Observer，直到调用到我们创建的Observer对象，到此onNext的流程执行完毕
妈耶，第一次写这么多字，按照示例分析完onNext()之后，对RxJava的执行流程有了小小的认识，收获很是挺多的，弄明白为啥subscribeOn只调用多次无用，子线程和主线程是怎样切换的，操作符是怎样转化数据的，目前水平有限，需要提高的地方很多，大佬们多多指教，抱拳！！
