Android进阶知识：RxJava相关
1. 前言
RxJava核心功能是一个用来完成异步操作的库，相对于其它异步操作的方法，RxJava的API使用更加的简洁。并且RxJava中还提供了很多功能强大的操作符，帮助我们解决很多原本复杂繁琐的代码逻辑，提高了代码质量。RxJava的实现是基于观察者模式，观察者模式中以下有三个比较重要的概念：
- 被观察者（Observable）
- 观察者（Observer）
- 订阅（subscribe）
被观察者是事件的发起者，被观察者与观察者建立订阅关系后，被观察者发送事件，观察者才能接收到事件。
2. 基础使用
RxJava的基础使用也很简单，分为三个步骤，分别是创建被观察者，创建观察者和建立订阅关系，具体代码如下。
// 1\. 创建被观察者
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(getClass().getName(), Thread.currentThread().getName() + " ObservableOnSubscribe subscribe");
emitter.onNext("string1");
emitter.onNext("string2");
emitter.onNext("string3");
emitter.onComplete();
}
});
// 2\. 创建观察者
Observer<String> observer = new Observer<String>() {
@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");
}
};
Log.d(getClass().getName(), Thread.currentThread().getName() + " observable:"+observable.getClass().getName());
// 3\. 建立订阅关系
observable.subscribe(observer);
运行日志：
3. 订阅源码流程
本文中所有源码基于RxJava2的2.2.11版本。首先来看看这个基本的订阅流程源码是怎么实现的。
3.1 创建被观察者
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
使用RxJava可以通过Observable的create方法创建一个被观察者对象。create方法从参数中传入一个ObservableOnSubscribe类型的source，然后方法中先校验了source是否为空，接着将传入的source封装成一个ObservableCreate对象，然后调用了RxJavaPlugins.onAssembly方法返回创建的好的Observable。接着进入onAssembly方法查看。
public static <T> Observable<T> onAssembly(@NonNull Observable<T> 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<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
......
}
3.2 创建观察者
public interface Observer<T> {
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<T> parent = new CreateEmitter<T>(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<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> 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<String> observable0 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(getClass().getName(), Thread.currentThread().getName() + " ObservableOnSubscribe subscribe");
emitter.onNext("string1");
emitter.onNext("string2");
emitter.onNext("string3");
emitter.onComplete();
}
});
Observer<String> observer = new Observer<String>() {
@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<String> observable1 = observable0.subscribeOn(Schedulers.newThread());
Log.d(getClass().getName(), Thread.currentThread().getName() + " observable1:"+observable1.getClass().getName());
Observable<String> 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<String> observable1 = observable0.subscribeOn(Schedulers.newThread());
Log.d(getClass().getName(), Thread.currentThread().getName() + " observable1:"+observable1.getClass().getName());
observable1.subscribe(observer);
运行结果：
先只调用subscribeOn方法运行查看结果，发现不仅被观察者发射事件运行在了子线程，观察者接收事件也运行在子线程，那么进入subscribeOn方法查看它的实现。
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(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<T> parent = new SubscribeOnObserver<T>(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<T> parent;
SubscribeTask(SubscribeOnObserver<T> 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<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
private static final long serialVersionUID = 8094547886072529208L;
final Observer<? super T> downstream;
final AtomicReference<Disposable> upstream;
SubscribeOnObserver(Observer<? super T> downstream) {
this.downstream = downstream;
this.upstream = new AtomicReference<Disposable>();
}
@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<Scheduler> {
@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<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;
}
scheduleActual方法里看到又将decoratedRun和DisposableContainer封装成ScheduledRunnable最后将这个ScheduledRunnable 交给构造函数中创建的线程池去运行，最终就会执行到前面看过的SubscribeTask中的run方法完成订阅逻辑，调用观察者的onNext等方法。到这里就看出最终的source.subscribe是会通过线程池切换到子线程中去执行了。
通过查看subscribeOn方法源码可以发现，方法里实际上是在前一个创建的ObservableCreate外面包了一层，把它包成一个ObservableSubscribeOn对象，同样的原先的Observer也被包了一层包成一个SubscribeOnObserver对象，而线程切换的工作是由Scheduler完成的。
4.2 observeOn
接着再来看看切换回主线程的方法observeOn，还是先修改使用代码，查看运行日志。
Observable<String> observable2 = observable0.observeOn(AndroidSchedulers.mainThread());
Log.d(getClass().getName(), Thread.currentThread().getName() + " observable2:"+observable2.getClass().getName());
observable2.subscribe(observer);
运行日志：
接着还是进入来看源码。
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
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));
}
这里看到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<T>(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<Scheduler>() {
@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<T>(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<T> parent = new CreateEmitter<T>(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);
}
}
查看ObserveOnObserver中的代码会发现onNext方法中先将传入的参数放入了一个队列，然后无论是onNext还是onComplete方法最后都调用了schedule方法，进而再进入查看，发现schedule方法中又调用了worker.schedule方法。这里的worker就是之前创建的HandlerWorker，这时再来看它的schedule方法。
public Disposable schedule(@NonNull Runnable run) {
return schedule(run, 0L, TimeUnit.NANOSECONDS);
}
单个参数schedule方法是在其父类中的，而这个方法中又调用另一个三个参数的schedule方法，这个方法父类中是抽象方法所以实现就在子类HandlerWorker里了。
@Override
@SuppressLint("NewApi") // Async will only be true when the API is available to call.
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
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
// 创建消息，并将主线程Handler和ScheduledRunnable
Message message = Message.obtain(handler, scheduled);
message.obj = this;
// 判断设置异步消息
if (async) {
message.setAsynchronous(true);
}
// 发送消息执行callback
handler.sendMessageDelayed(message, unit.toMillis(delay));
// 检查是否取消订阅
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
在子类的这个方法里在做了取消订阅的判断后将方法传入的Runnable和Handler又封装到一个ScheduledRunnable对象中。接着创建了一个Message并将ScheduledRunnable放入Message，最后调用handler.sendMessageDelayed方法通过这个主线程的Handler执行这个ScheduledRunnable。
最后来追溯下ScheduledRunnable到底执行了什么，不过猜也知道最后一定调用到观察者中的对应方法。
private static final class ScheduledRunnable implements Runnable, Disposable {
private final Handler handler;
private final Runnable delegate;
private volatile boolean disposed; // Tracked solely for isDisposed().
ScheduledRunnable(Handler handler, Runnable delegate) {
this.handler = handler;
this.delegate = delegate;
}
@Override
public void run() {
try {
delegate.run();
} catch (Throwable t) {
RxJavaPlugins.onError(t);
}
}
......
}
ScheduledRunnable中的run方法很简单就是调用了构造中传入的Runnable的run方法。而根据之前看过得创建ScheduledRunnable时传入的Runnable又是从scheduleDirect方法中传入的，而scheduleDirect方法中的Runnable又是从worker.schedule(this)方法时传入的，根据上下文代码发现这个this指代的是ObserveOnObserver对象，于是进一步进入它的run方法查看。
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
......
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.downstream = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
......
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
......
}
可以看到run方法中判断了outputFused的真假，然后分别调用了drainFused和drainNormal方法。这里的outputFused是与RxJava2中的背压处理相关暂时先不管，根据方法名也能知道正常调用会执行drainNormal方法，于是直接来看drainNormal方法。
void drainNormal() {
int missed = 1;
// 存放onNext传入的事件对象队列
final SimpleQueue<T> q = queue;
// 传入的观察者对象
final Observer<? super T> a = downstream;
// 循环check事件是否完成或者发生错误
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;
}
// 判断队列中取出的发送事件传入的对象v是否为空
if (empty) {
break;
}
// 执行观察者对象的onNext方法
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
drainNormal方法中先通过checkTerminated方法校验发送事件是否完成或者发生异常，接着从队列中取出事件对象，再次判断是否完成或者发生错误和取出的对象是否为空，没有问题的话就会执行观察者的onNext方法。而发送完成和出现异常的方法则是在checkTerminated方法处理。
boolean checkTerminated(boolean d, boolean empty, Observer<? super T> a) {
if (disposed) {
queue.clear();
return true;
}
if (d) {
Throwable e = error;
if (delayError) {
if (empty) {
disposed = true;
if (e != null) {
a.onError(e);
} else {
a.onComplete();
}
worker.dispose();
return true;
}
} else {
if (e != null) {
disposed = true;
queue.clear();
a.onError(e);
worker.dispose();
return true;
} else
if (empty) {
disposed = true;
a.onComplete();
worker.dispose();
return true;
}
}
}
return false;
}
在checkTerminated方法里根据delayError判断是否设置了超时的错误，接着再根据获得的错误e是否为空再决定调用的是观察者的onError()方法还是onComplete方法。至此observeOn切换线程的流程也梳理结束。
5. map操作符
RxJava中有很多功能强大的操作符，通过使用这些操作符，可以很容易的解决代码编写时遇到的一些复杂繁琐的问题。这里就用map操作符来作为一个例子，来看看操作符是怎样工作的。首先还是来了解map操作符的使用方法和作用。
final Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(getClass().getName(), Thread.currentThread().getName() + " ObservableOnSubscribe subscribe");
emitter.onNext("5");
emitter.onComplete();
}
});
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(getClass().getName(), Thread.currentThread().getName() + " onSubscribe");
}
@Override
public void onNext(Integer i) {
Log.d(getClass().getName(), Thread.currentThread().getName() + " onNext "+i);
}
@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<Integer> mapObservable = observable.map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return Integer.parseInt(s);
}
});
Log.d(getClass().getName(), Thread.currentThread().getName() + " mapObservable:"+mapObservable.getClass().getName());
mapObservable.subscribe(observer);
运行日志：
map操作符作用是可以将被观察者发送事件的数据类型转换成其他的数据类型。它的使用方法很简单，例如上面这个例子就将一开始发送的String类型转换成观察者接收到的Integer类型。下面开始看map方法的源码。
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
看到map方法中依旧还是同样的套路，通过RxJavaPlugins.onAssembly方法返回一个被观察者对象，只不过这次构建传入的类型又是另一个ObservableMap类型的对象。订阅的流程前面已经看过了，这里和之前的一样最终会走到ObservableMap的subscribeActual方法，所以直接来看这个方法。
@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
ObservableMap的subscribeActual方法里看到很熟悉还是会调用source.subscribe方法，只是这里传入的Observer对象是一个MapObserver对象。接下来的逻辑又和之前一样，根据之前的经验source.subscribe方法最终会调用Observer的onNext方法，所以接下来直接来看MapObserver的onNext方法。
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
downstream.onNext(v);
}
MapObserver的onNext方法里的逻辑很简单，在做了一些的判断后调用mapper.apply(t)方法获得类型转换后的事件传递对象，最后就会调用观察者的downstream.onNext方法，这里的downstream就是订阅方法传入的观察者对象。跟踪mapper可以找到，它是从MapObserver构造时传入的一个Function类型，即是在使用map操作符时传入的那个Function对象，又因为在使用时实现了Function的apply方法完成了数据的类型转换逻辑，所以这里调用mapper.apply(t)方法就可以获得到转换后的数据。
6. 总结
以上就是关于RxJava源码工作流程的相关总结，总而言之，观察者模式还是其核心设计思想。除此之外，通过源码阅读还发现，无论在线程切换方面还是其它功能的操作符的实现，根本上来说都是在其原有的被观察者或观察者基础上包装成一个新的对象，功能逻辑由新对象中的方法来实现完成。
喜欢本文的话，不妨顺手给我点个小赞、评论区留言或者转发支持一下呗😜😜😜~
点击【GitHub】还有彩蛋哦！！！
