public static Observable onAssembly(@NonNull Observable source) {

Function<? super Observable, ? extends Observable> f = onObservableAssembly;

if (f != null) {

return apply(f, source);

}

return source;

}

onAssembly方法中首先是一个Hook实现，这里可以理解为一个代理。可以看到这里先判断onObservableAssembly是否为空，为空就直接返回传入的source，否则再调用apply方法。这里可以继续跟踪一下onObservableAssembly。

@SuppressWarnings(“rawtypes”)

@Nullable

static volatile Function<? super Observable, ? extends Observable> onObservableAssembly;

/**

• Sets the specific hook function.

• @param onObservableAssembly the hook function to set, null allowed

*/

@SuppressWarnings(“rawtypes”)

public static void setOnObservableAssembly(@Nullable Function<? super Observable, ? extends Observable> onObservableAssembly) {

if (lockdown) {

throw new IllegalStateException(“Plugins can’t be changed anymore”);

}

RxJavaPlugins.onObservableAssembly = onObservableAssembly;

}

它是RxJavaPlugins中的成员变量，默认为空，并且提供了一个set方法来设置它。因为默认为空，所以默认返回的就是传入的source。这里的代理默认是不会对Observable做什么操作，如果需要有特殊的需求可以调用set方法实现自己的代理。而默认返回的source类型为ObservableCreate对象也实现了Observable接口。

public final class ObservableCreate extends Observable {

final ObservableOnSubscribe source;

public ObservableCreate(ObservableOnSubscribe source) {

this.source = source;

}

…

}

3.2 创建观察者

public interface Observer {

void onSubscribe(@NonNull Disposable d);

void onNext(@NonNull T t);

void onError(@NonNull Throwable e);

void onComplete();

}

观察者Observer是一个接口，其中提供了一些方法，使用时创建接口的实现，并根据需求在方法中做自己的实现。

3.3 建立订阅关系

建立订阅关系调用了Observable的subscribe方法。

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.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins”);

subscribeActual(observer);

} catch (NullPointerException e) { // NOPMD

…

} catch (Throwable e) {

…

}

}

方法中还是先判断了传入参数observer是否为空，接着还是一个Hook实现，这里就不细究了，获得Hook返回的observer后再次判断是否为空，之后调用了subscribeActual方法。

protected abstract void subscribeActual(Observer<? super T> observer);

Observable的subscribeActual方法是个抽象方法，之前看过这里的Observable实际实现是个ObservableCreate对象，所以再进入ObservableCreate类查看对应方法。

@Override

protected void subscribeActual(Observer<? super T> observer) {

CreateEmitter parent = new CreateEmitter(observer);

observer.onSubscribe(parent);

try {

source.subscribe(parent);

} catch (Throwable ex) {

Exceptions.throwIfFatal(ex);

parent.onError(ex);

}

}

ObservableCreate中的subscribeActual方法中先创建了一个CreateEmitter发射器对象，并将observer对象传入。接着调用了observer的onSubscribe方法，此时观察者的onSubscribe方法执行。最后调用了source的subscribe方法。

Observable observable = Observable.create(new ObservableOnSubscribe() {

@Override

public void subscribe(ObservableEmitter emitter) throws Exception {

Log.d(getClass().getName(), Thread.currentThread().getName() + " ObservableOnSubscribe subscribe");

emitter.onNext(“string1”);

emitter.onNext(“string2”);

emitter.onNext(“string3”);

emitter.onComplete();

}

});

这个source就是在create方法中传入的ObservableOnSubscribe。它的subscribe方法中通过调用ObservableEmitter的方法发送事件，这里的ObservableEmitter就是之前创建的CreateEmitter对象，所以再来进一步看看它其中的方法。

CreateEmitter(Observer<? super T> observer) {

this.observer = observer;

}

@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);

}

}

@Override

public void onComplete() {

if (!isDisposed()) {

try {

observer.onComplete();

} finally {

dispose();

}

}

}

CreateEmitter的构造函数接收了观察者对象，然后在调用onNext方法时先做了空判断，再对isDisposed进行取消订阅的判断，之后调用了observer的onNext方法，也就是观察者的onNext方法。同样的onComplete中最终也是调用了observer的onComplete方法。至此RxJava中的基本订阅流程的源码就梳理完了。

4. 线程切换

---

RxJava中有个很重要的功能，就是能方便的切换线程，来看下它的使用，还是之前基础使用中的例子进行修改。

Observable observable0 = Observable.create(new ObservableOnSubscribe() {

@Override

public void subscribe(ObservableEmitter emitter) throws Exception {

Log.d(getClass().getName(), Thread.currentThread().getName() + " ObservableOnSubscribe subscribe");

emitter.onNext(“string1”);

emitter.onNext(“string2”);

emitter.onNext(“string3”);

emitter.onComplete();

}

});

Observer observer = new Observer() {

@Override

public void onSubscribe(Disposable d) {

Log.d(getClass().getName(), Thread.currentThread().getName() + " onSubscribe");

}

@Override

public void onNext(String s) {

Log.d(getClass().getName(), Thread.currentThread().getName() + " onNext "+s);

}

@Override

public void onError(Throwable e) {

Log.d(getClass().getName(), Thread.currentThread().getName() + " onError");

}

@Override

public void onComplete() {

Log.d(getClass().getName(), Thread.currentThread().getName() + " onComplete");

}

};

Observable observable1 = observable0.subscribeOn(Schedulers.newThread());

Log.d(getClass().getName(), Thread.currentThread().getName() + " observable1:"+observable1.getClass().getName());

Observable observable2 = observable1.observeOn(AndroidSchedulers.mainThread());

Log.d(getClass().getName(), Thread.currentThread().getName() + " observable2:"+observable2.getClass().getName());

observable2.subscribe(observer);

被观察者和观察者的创建和之前一样，在建立订阅关系时调用subscribeOn和observeOn方法进行线程的切换。这里每个方法返回的都是Observable类型，所以可以采用链式调用，这也是RxJava的一个特点，但是这里没有采用这种写法，而是将其拆分开来写并且日志打印出每个Observable的具体类型，这是为了方便之后源码理解。 运行结果日志：

4.1 subscribeOn

Observable observable1 = observable0.subscribeOn(Schedulers.newThread());

Log.d(getClass().getName(), Thread.currentThread().getName() + " observable1:"+observable1.getClass().getName());

observable1.subscribe(observer);

运行结果：

先只调用subscribeOn方法运行查看结果，发现不仅被观察者发射事件运行在了子线程，观察者接收事件也运行在子线程，那么进入subscribeOn方法查看它的实现。

public final Observable subscribeOn(Scheduler scheduler) {

ObjectHelper.requireNonNull(scheduler, “scheduler is null”);

return RxJavaPlugins.onAssembly(new ObservableSubscribeOn(this, scheduler));

}

可以看到subscribeOn方法和subscribe方法有些类似。首先是判断传入的scheduler是否为空，然后同样调用RxJavaPlugins.onAssembly方法，这次构建了一个ObservableSubscribeOn对象返回。而subscribeOn方法之后还是调用了subscribe方法，根据之前的分析，subscribe方法最终会调用到subscribeActual方法，不过此时的subscribeActual方法不再是ObservableCreate中的而是ObservableSubscribeOn中的subscribeActual方法。

@Override

public void subscribeActual(final Observer<? super T> observer) {

final SubscribeOnObserver parent = new SubscribeOnObserver(observer);

observer.onSubscribe(parent);

parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));

}

ObservableSubscribeOn的subscribeActual方法中流程和之前的也很类似，这次是先创建了一个SubscribeOnObserver对象，将观察者对象传入，接着同样先调用了observer.onSubscribe方法，然后将传入的SubscribeOnObserver封装入了一个SubscribeTask对象中，接着调用了scheduler.scheduleDirect方法再将返回结果得到的Disposable设置到SubscribeOnObserver中。下面一个方法一个方法看。首先是创建SubscribeTask。

final class SubscribeTask implements Runnable {

private final SubscribeOnObserver parent;

SubscribeTask(SubscribeOnObserver parent) {

this.parent = parent;

}

@Override

public void run() {

source.subscribe(parent);

}

}

SubscribeTask是ObservableSubscribeOn的内部类，其实现很简单就是实现了一个Runnable接口，构造方法中传入了SubscribeOnObserver对象，在其run方法中调用了ObservableSubscribeOn中的成员变量source的subscribe方法。这个source是在创建ObservableSubscribeOn时传入的，根据前面的代码可以找到是在subscribeOn方法中创建的对象并且这个source对应传入的是当前这个Observable对象即通过Observable.create获得的被观察者对象，其实现之前看过是一个ObservableCreate所以这里就和之前一样又会走到了其父类Observable的subscribe方法中，继而调用ObservableCreate的subscribeActual方法，之后最终会调用到观察者的对应onNext等方法，不过此时的观察者不直接是在使用时创建传入的Observer，而是之前看到的SubscribeOnObserver类型，不过其中的onNext等方法还是调用了在使用时创建传入的Observer的对应方法。

static final class SubscribeOnObserver extends AtomicReference implements Observer, Disposable {

private static final long serialVersionUID = 8094547886072529208L;

final Observer<? super T> downstream;

final AtomicReference upstream;

SubscribeOnObserver(Observer<? super T> downstream) {

this.downstream = downstream;

this.upstream = new AtomicReference();

}

@Override

public void onNext(T t) {

downstream.onNext(t);

}

@Override

public void onError(Throwable t) {

downstream.onError(t);

}

@Override

public void onComplete() {

downstream.onComplete();

}

…

}

下面接着看到scheduleDirect这个方法，在创建好SubscribeTask之后调用了scheduleDirect方法。这里的scheduler就是subscribeOn中传入的,对应开始例子中的Schedulers.newThread。

public static Scheduler newThread() {

return RxJavaPlugins.onNewThreadScheduler(NEW_THREAD);

}

// 静态成员变量NEW_THREAD

static final Scheduler NEW_THREAD;

NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());

进入Schedulers.newThread一步步跟踪，看到newThread方法返回静态成员变量中的NEW_THREAD，而NEW_THREAD又是通过NewThreadTask创建。

static final class NewThreadTask implements Callable {

@Override

public Scheduler call() throws Exception {

return NewThreadHolder.DEFAULT;

}

}

static final Scheduler DEFAULT = new NewThreadScheduler();

继续跟踪查看发现NewThreadTask实际是实现了Callable接口，其call方法中返回了静态内部类中的NewThreadHolder.DEFAULT。这个DEFAULT的实现类型为NewThreadScheduler。至此终于找到了我们传入的Scheduler的真正实现类。于是继续看其scheduleDirect方法。

public Disposable scheduleDirect(@NonNull Runnable run) {

return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);

}

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);

return task;

}

scheduleDirect方法是在其父类中实现的，看到其中进而调用了同名重载方法，方法中首先是调用createWorker方法创建一个Worker。这个方法的实现就是在NewThreadScheduler中了。

public Worker createWorker() {

return new NewThreadWorker(threadFactory);

}

createWorker方法中只做了一件事就是创建返回了一个NewThreadWorker。

public class NewThreadWorker extends Scheduler.Worker implements Disposable {

private final ScheduledExecutorService executor;

volatile boolean disposed;

public NewThreadWorker(ThreadFactory threadFactory) {

executor = SchedulerPoolFactory.create(threadFactory);

}

…

}

NewThreadWorker中看到创建了一个线程池，再回到scheduleDirect方法，创建完Worker后将传入的Runnable即SubscribeTask进行一个装饰得到新的Runnable对象。接着将Worker和新的Runnable封装到一个DisposeTask对象中。

static final class DisposeTask implements Disposable, Runnable, SchedulerRunnableIntrospection {

@NonNull

final Runnable decoratedRun;

@NonNull

final Worker w;

@Nullable

Thread runner;

DisposeTask(@NonNull Runnable decoratedRun, @NonNull Worker w) {

this.decoratedRun = decoratedRun;

this.w = w;

}

@Override

public void run() {

runner = Thread.currentThread();

try {

decoratedRun.run();

} finally {

dispose();

runner = null;

}

}

…

}

DisposeTask同样实现了Runnable接口，在run方法中调用了从构造传入的decoratedRun的run方法执行任务。回到最后一步，调用Worker的schedule方法，这里就对应的NewThreadWorker的schedule方法。

public Disposable schedule(@NonNull final Runnable action, long delayTime, @NonNull TimeUnit unit) {

if (disposed) {

return EmptyDisposable.INSTANCE;

}

return scheduleActual(action, delayTime, unit, null);

}

schedule方法中又进一步调用了其scheduleActual方法

public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {

Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

if (parent != null) {

if (!parent.add(sr)) {

return sr;

}

}

Future<?> f;

try {

if (delayTime <= 0) {

f = executor.submit((Callable)sr);

} else {

f = executor.schedule((Callable)sr, delayTime, unit);

}

sr.setFuture(f);

} catch (RejectedExecutionException ex) {

if (parent != null) {

parent.remove(sr);

}

RxJavaPlugins.onError(ex);

}

return sr;

}

scheduleActual方法里看到又将decoratedRun和DisposableContainer封装成ScheduledRunnable最后将这个ScheduledRunnable 交给构造函数中创建的线程池去运行，最终就会执行到前面看过的SubscribeTask中的run方法完成订阅逻辑，调用观察者的onNext等方法。到这里就看出最终的source.subscribe是会通过线程池切换到子线程中去执行了。

通过查看subscribeOn方法源码可以发现，方法里实际上是在前一个创建的ObservableCreate外面包了一层，把它包成一个ObservableSubscribeOn对象，同样的原先的Observer也被包了一层包成一个SubscribeOnObserver对象，而线程切换的工作是由Scheduler完成的。

4.2 observeOn

接着再来看看切换回主线程的方法observeOn，还是先修改使用代码，查看运行日志。

Observable observable2 = observable0.observeOn(AndroidSchedulers.mainThread());

Log.d(getClass().getName(), Thread.currentThread().getName() + " observable2:"+observable2.getClass().getName());

observable2.subscribe(observer);

运行日志：

接着还是进入来看源码。

public final Observable observeOn(Scheduler scheduler) {

return observeOn(scheduler, false, bufferSize());

}

public final Observable observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {

ObjectHelper.requireNonNull(scheduler, “scheduler is null”);

ObjectHelper.verifyPositive(bufferSize, “bufferSize”);

return RxJavaPlugins.onAssembly(new ObservableObserveOn(this, scheduler, delayError, bufferSize));

}

这里看到observeOn方法里调用了重载方法，方法中还是同一个套路，不过这里创建的又是另一个对象ObservableObserveOn了。根据前面的经验这里就又是将前一个Observable传递到ObservableObserveOn中的成员变量source上，这里看到就是构造函数中的第一个参数。接着还是会调用subscribe与观察者建立订阅关系进而会执行到ObservableObserveOn对象的subscribeActual方法。

@Override

protected void subscribeActual(Observer<? super T> observer) {

if (scheduler instanceof TrampolineScheduler) {

source.subscribe(observer);

} else {

Scheduler.Worker w = scheduler.createWorker();

source.subscribe(new ObserveOnObserver(observer, w, delayError, bufferSize));

}

}

subscribeActual方法中判断了scheduler的类型，这里的scheduler就是由AndroidSchedulers.mainThread()传入的，于是先来看一下这个方法。

public static Scheduler mainThread() {

return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);

}

private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(

new Callable() {

@Override public Scheduler call() throws Exception {

return MainHolder.DEFAULT;

}

});

private static final class MainHolder {

static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()), false);

}

从mainThread开始看，发现代码调用逻辑和之前的Schedulers.newThread方法类似，最终会返回一个HandlerScheduler而这个Scheduler中的Handler则是主线程的Handler，看到这里就能猜想到了，后面观察者的对应方法一定是由这个Handler来切换到主线程执行的。回到subscribeActual方法。

@Override

protected void subscribeActual(Observer<? super T> observer) {

if (scheduler instanceof TrampolineScheduler) {

source.subscribe(observer);

} else {

Scheduler.Worker w = scheduler.createWorker();

source.subscribe(new ObserveOnObserver(observer, w, delayError, bufferSize));

}

}

这里判断完类型会走else中的方法首先还是会调用HandlerScheduler的createWorker方法创建一个Worker。

@Override

public Worker createWorker() {

return new HandlerWorker(handler, async);

}

这里是个HandlerWorker其中具体方法后面再看。接着上面创建完Worker后同样还是一样调用source.subscribe创建了一个ObserveOnObserver对象传入。这里的source就还是之前的ObservableCreate，所以这里还是会调用ObservableCreate中的subscribeActual方法。

protected void subscribeActual(Observer<? super T> observer) {

CreateEmitter parent = new CreateEmitter(observer);

observer.onSubscribe(parent);

try {

source.subscribe(parent);

} catch (Throwable ex) {

Exceptions.throwIfFatal(ex);

parent.onError(ex);

}

}

ObservableCreate中的subscribeActual方法中的逻辑之前看过，不过此时传入的observer仍然不再是在使用时创建的观察者对象了，而是传过来的ObserveOnObserver对象，此时创建的CreateEmitter中的observer也就是这个ObserveOnObserver对象。和之前逻辑一样，接着就会调用observer的onNext等方法，此时调用的即是ObserveOnObserver中的onNext等方法。所以进入ObserveOnObserver查看。

@Override

public void onNext(T t) {

if (done) {

return;

}

if (sourceMode != QueueDisposable.ASYNC) {

queue.offer(t);

}

schedule();

}

@Override

public void onComplete() {

if (done) {

return;

}

done = true;

schedule();

}

void schedule() {

if (getAndIncrement() == 0) {

worker.schedule(this);

结尾

我还总结出了互联网公司Android程序员面试涉及到的绝大部分面试题及答案，并整理做成了文档，以及系统的进阶学习视频资料，免费分享给大家。
（包括Java在Android开发中应用、APP框架知识体系、高级UI、全方位性能调优，NDK开发，音视频技术，人工智能技术，跨平台技术等技术资料），希望能帮助到你面试前的复习，且找到一个好的工作，也节省大家在网上搜索资料的时间来学习。

领取方式：关注+点赞+点击我的GitHub 免费获取！

ex);

parent.onError(ex);

}

}

ObservableCreate中的subscribeActual方法中的逻辑之前看过，不过此时传入的observer仍然不再是在使用时创建的观察者对象了，而是传过来的ObserveOnObserver对象，此时创建的CreateEmitter中的observer也就是这个ObserveOnObserver对象。和之前逻辑一样，接着就会调用observer的onNext等方法，此时调用的即是ObserveOnObserver中的onNext等方法。所以进入ObserveOnObserver查看。

@Override

public void onNext(T t) {

if (done) {

return;

}

if (sourceMode != QueueDisposable.ASYNC) {

queue.offer(t);

}

schedule();

}

@Override

public void onComplete() {

if (done) {

return;

}

done = true;

schedule();

}

void schedule() {

if (getAndIncrement() == 0) {

worker.schedule(this);

结尾

我还总结出了互联网公司Android程序员面试涉及到的绝大部分面试题及答案，并整理做成了文档，以及系统的进阶学习视频资料，免费分享给大家。
（包括Java在Android开发中应用、APP框架知识体系、高级UI、全方位性能调优，NDK开发，音视频技术，人工智能技术，跨平台技术等技术资料），希望能帮助到你面试前的复习，且找到一个好的工作，也节省大家在网上搜索资料的时间来学习。

领取方式：关注+点赞+点击我的GitHub 免费获取！