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 免费获取！
