rxjava代码
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("有情况");
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
Log.e("qwer", s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
然后create和subscribe也不讲了（可以看前面文章）
1、直接看observeOn
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
再进入重载的observeOn方法
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
再进入ObservableObserveOn
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
同样的套路，完成赋值
通过前面的三篇文章，我们已经知道接下来会进入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));
}
}
很显然，scheduler 不是 TrampolineScheduler类型，也就是进入else代码中，调用了scheduler.createWorker();
2、这个时候必须要先具体了解一下scheduler了
这个scheduler就是我们传的AndroidSchedulers.mainThread()
进入该方法
public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
其实这里onMainThreadScheduler返回的就是本身
3、所以继续看MAIN_THREAD
private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
new Callable<Scheduler>() {
@Override public Scheduler call() throws Exception {
return MainHolder.DEFAULT;
}
});
进入RxAndroidPlugins.initMainThreadScheduler
public static Scheduler initMainThreadScheduler(Callable<Scheduler> scheduler) {
if (scheduler == null) {
throw new NullPointerException("scheduler == null");
}
Function<Callable<Scheduler>, Scheduler> f = onInitMainThreadHandler;
if (f == null) {
return callRequireNonNull(scheduler);
}
return applyRequireNonNull(f, scheduler);
}
因为f为空（为什么为空，就是我没有对onInitMainThreadHandler进行赋值），所以返回了callRequireNonNull(scheduler)，进入该方法
static Scheduler callRequireNonNull(Callable<Scheduler> s) {
try {
Scheduler scheduler = s.call();
if (scheduler == null) {
throw new NullPointerException("Scheduler Callable returned null");
}
return scheduler;
} catch (Throwable ex) {
throw Exceptions.propagate(ex);
}
}
其实就是返回了s.call()，而s.call()是什么，不错，就是步骤3最开始的
return MainHolder.DEFAULT;
4、继续查看 MainHolder.DEFAULT
private static final class MainHolder {
static final Scheduler DEFAULT
= new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
}
！！！发现重点
这里返回了一个HandlerScheduler，构造函数参数还放了一个
new Handler(Looper.getMainLooper())，是不是感觉有点感觉
进入HandlerScheduler
HandlerScheduler(Handler handler, boolean async) {
this.handler = handler;
this.async = async;
}
只是赋值，一个主线程handler,然后async是false
而这个时候我们联系步骤1最后是不是应该看它的createWorker方法
public Worker createWorker() {
return new HandlerWorker(handler, async);
}
好，所以此时回到步骤1的最后，Scheduler.Worker就是HandlerWorker
5、顺着步骤1最后的else代码继续看
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
source.subscribe就不用讲了，直接看ObserveOnObserver
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.downstream = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
一系列的赋值操作
嗯哼？是不是好像结束了？？
不存在的！！！
注意它可是Observer，注意它的方法
6、首先看onSubscribe方法，因为它会被第一个调用（看过第一篇文章的都知道），它也是我们自己new的观察者的第一个回调方法
public void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
this.upstream = d;
if (d instanceof QueueDisposable) {
@SuppressWarnings("unchecked")
QueueDisposable<T> qd = (QueueDisposable<T>) d;
int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
if (m == QueueDisposable.SYNC) {
sourceMode = m;
queue = qd;
done = true;
downstream.onSubscribe(this);
schedule();
return;
}
if (m == QueueDisposable.ASYNC) {
sourceMode = m;
queue = qd;
downstream.onSubscribe(this);
return;
}
}
queue = new SpscLinkedArrayQueue<T>(bufferSize);
downstream.onSubscribe(this);
}
}
根据之前的代码我们知道这个d不是QueueDisposable
所以直接跳过if
对queue 进行了初始化
这里queue 说一下，虽然不知道具体细节，单可以肯定的是，他是一个队列
然后继续往下看downstream.onSubscribe(this)完成Disposable的继续传递
7、接下来再看什么？是不是就是我们的onNext()方法了
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
因为我们知道这个t肯定要往下传，所以done为false,sourceMode 也不等于QueueDisposable.ASYNC
然后我们被观察者发送的数据t就被压入了队列queue里去了
8、然后执行schedule()
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
这里getAndIncrement()应该是线程任务数（我猜的，返回这里肯定要为0）
然后进入schedule方法
public Disposable schedule(@NonNull Runnable run) {
return schedule(run, 0L, TimeUnit.NANOSECONDS);
}
走了重载的方法，而且schedule是个抽象方法
9、那此时你还记得这个worker是谁了么？
不错，看步骤4的最后，就是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);
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;
}
在代码里只需要注意重点
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run)
这个handler是主线程handler(步骤4)
这个run就是步骤5里new的那个ObserveOnObserver，它实现了Runnable接口
下面执行代码
handler.sendMessageDelayed(message, unit.toMillis(delay))
好，此时run线程被执行（如果不知道为什么可以好好看看handler基础），而且是在主线程执行！！！说明线程已经切换到主线程了
那么接下来呢？
不错！回到步骤5里new的那个ObserveOnObserver的run方法
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
这里outputFused为false（赋值true的地方我没有执行，其实具体看drainFused和drainNormal两个方法的代码，也可以辨别出其为false）
10、然后进入drainNormal方法
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = downstream;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
disposed = true;
upstream.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;
}
}
}
直接说重点
首先开启for循环，当队列不为空的时候
v = q.poll();
拿到队列里的值（就是步骤7里offer的那个）
然后a.onNext(v)完成事件传递（往上面看，a就是downstream）
自此就算结束了
总结---
关键步骤在9，通过主线程的hanlder完成线程切换
