前言
上回讲到subscribe()订阅观察者，这回咱们来聊聊subscribeOn()方法，官方解释是：在指定的线程中订阅该被观察者，你肯定知道subscribeOn调用多次只有最上面的那一次是生效的，真的是这样吗？留着这个问题，start read the fucking code~~
1.示例
Observable.create(...)
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "accept: first Thread = " + Thread.currentThread().getName());
}
})
.subscribeOn(Schedulers.computation())
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "accept: second Thread = " + Thread.currentThread().getName());
}
})
.subscribeOn(Schedulers.io())
.subscribe(new Observer<Integer>() {...}
});
//日志
D: onSubscribe: Thread = main
D: accept: first Thread = RxComputationThreadPool-2
D: accept: second Thread = RxComputationThreadPool-2
D: onNext: Thread = RxComputationThreadPool-2
D: onComplete: Thread = RxComputationThreadPool-2
我们先不关心just,doOnNext操作符，这里仅仅是用来打印日志，结合上回的分析开始本次源码分析~
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
//懂的自然懂~，这里直接返回ObservableSubscribeOn实例
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
//持有上游的Source对象，也就是ObservableCreate被观察者对象
super(source);
//持有我们传入的线程调度对象
this.scheduler = scheduler;
}
}
2.开始订阅
有了上篇的分析，我们知道这里直接调用的是ObservableSubscribeOn的subscribeActual()方法
public void subscribeActual(final Observer<? super T> observer) {
//实例化SubscribeOnObserver观察者对象，传入observer进行持有
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
//打印第一行日志 ---onSubscribe: Thread = main
//因为当前线程是UI线程，所以是线程名为main
observer.onSubscribe(parent);
//这里没有直接订阅，而是先进行了线程切换，才开始继续’向上订阅‘
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
//SubscribeTask是ObservableSubscribeOn的内部类，实现了Runnable接口，注定要在线程里完成使命
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
//持有我们生成的SubscribeOnObserver对象
this.parent = parent;
}
@Override
public void run() {
//source是指‘上游的’ObservableSource对象
source.subscribe(parent);
}
}
//接着再看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) {
//1.createWorker是Scheduler的抽象函数，我们外部传入的是Schedulers.io()对象，即IoScheduler实例
//这里的Worker是EventLoopWorker类实例，持有线程池之类的，这里不究细节
final Worker w = createWorker();
//属性的钩子函数
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
//DisposeTask实现Runnable，Disposable接口，进行一层封装
DisposeTask task = new DisposeTask(decoratedRun, w);
//将我们创建的task传进去
w.schedule(task, delay, unit);
return task;
}
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
//首先判断是否已经被安排了(取消订阅了)
if (tasks.isDisposed()) {
// don't schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}
//这里有很多线程池，Worker缓存池的东西，我们先不纠这些细节，跳过往下看scheduleActual()
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
//我也不知道，为什么要将SubscribeTask对象包裹这么多次。。。。。。
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}
Future<?> f;
try {
//延迟默认是0
if (delayTime <= 0) {
//最终看到这里，我们的SubscribeTask对象在线程池中终于被安排上了，好累~~
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;
}
3.总结
subscribeOn的职责
- 调用onSubscribe()开始订阅
- 切换指定线程线程
- 在指定线程中继续向上订阅
后续的向上订阅到发射数据，都是在指定的线程中执行，这样也解释了我们之前打印的日志，在UI线程中订阅，切换到io线程中继续向上订阅，再次切换到Computation线程向上订阅，后续的发射发射数据都是RxComputationThreadPool线程中，也证明了调用多次subscribeOn()只有最上面那个有效。
