通过 Scheduler 来控制被观察者在哪个线程发射,观察者在哪个线程接收。默认情况,发射时在哪个线程,接收就在哪个线程。
RxJava 内置了几个 Scheduler,通过 Schedulers 来获取。
Observable.create(ObservableOnSubscribe<String> { emitter ->
Log.e("RX", "发射线程 ${Thread.currentThread().name}")
emitter.onNext("a")
}).subscribeOn(Schedulers.io()) // 在 io 线程发射
.observeOn(Schedulers.computation()) // 在计算线程接收
.subscribe( {
Log.e("RX", "接收线程 ${Thread.currentThread().name}")
Log.e("RX", "接收数据 $it")
})
日志显示:
发射线程 RxCachedThreadScheduler-1
接收线程 RxComputationThreadPool-1
接收数据 a
如果 observeOn(Schedulers.trampoline())
,意思是在当前线程,由于发射时将线程切换到 io 上去了,所以接收时也在这个 io 线程上,日志显示:
发射线程 RxCachedThreadScheduler-1
接收线程 RxCachedThreadScheduler-1
接收数据 a
Observable.create(ObservableOnSubscribe<String> { emitter ->
Log.e("RX", "发射线程 ${Thread.currentThread().name}")
emitter.onNext("a")
}).subscribeOn(Schedulers.io()) // 在 io 线程发射
.observeOn(Schedulers.computation())
.map {
Log.e("RX", "第一次转换数据的线程 ${Thread.currentThread().name}")
"$it$it"
} // 双倍
.observeOn(Schedulers.newThread())
.map {
Log.e("RX", "第二次转换数据的线程 ${Thread.currentThread().name}")
"$it$it" } // 四倍
.observeOn(Schedulers.single())
.subscribe( {
Log.e("RX", "接收线程 ${Thread.currentThread().name}")
Log.e("RX", "接收数据 $it")
})
日志:
发射线程 RxCachedThreadScheduler-1
第一次转换数据的线程 RxComputationThreadPool-1
第二次转换数据的线程 RxNewThreadScheduler-1
接收线程 RxSingleScheduler-1
接收数据 aaaa
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
...
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
public final Observable<T> subscribeOn(Scheduler scheduler) {
...
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
...
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
基本上每调一个方法,就创建了一个 Observable 的实现类,然后将上层调用它的 Observable 作为自己内部的 source。
最后调用 ObservableObserveOn 的 subscribe,内部调用 subscribeActual,参数是用户传递进来的 Observer。
protected void subscribeActual(Observer<? super T> observer) {
// 如果是 TrampolineScheduler,在当前线程执行,不涉及任何线程的切换,所以直接调 source.subscribe
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
如果 scheduler 不是 TrampolineScheduler,比如 NewThreadScheduler,先调用 createWorker 方法,创建 NewThreadWorker,实现了 Disposable。
public class NewThreadWorker extends Scheduler.Worker implements Disposable {
private final ScheduledExecutorService executor;
public NewThreadWorker(ThreadFactory threadFactory) {
executor = SchedulerPoolFactory.create(threadFactory);
}
...
}
构造中创建一个线程池,再深入下去看,创建了一个只有一个核心线程的线程池对象。然后根据外界传入的 Observer 和这个线程池封装出另一个 Observer。所以在 ObservableSubscribeOn 的对象 source 上用重新封装好的一个观察者订阅它。这样就会调用 ObservableSubscribeOn 的 subscribeActual 方法。
@Override
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
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 是一个 Runnable,内部持有最后封装的 Observer。且 run 方法就是用这个观察者去订阅 source。在这个例子里,ObservableSubscribeOn 中的 source 就是上层 ObservableCreate。
而 scheduler.scheduleDirect 这个 Runnable,假设发射的 schedule 是 IoScheduler,scheduleDirect 最后是
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
// 创建线程,返回 EventLoopWorker
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
DisposeTask task = new DisposeTask(decoratedRun, w);
w.schedule(task, delay, unit);
return task;
}
看最后调用 EventLoopWorker 的 schedule 方法
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
...
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);
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;
}
这里使用线程池方法执行参数 run 里面的任务,参数这个 run 是 DisposeTask 对象。
public void run() {
runner = Thread.currentThread();
try {
decoratedRun.run();
} finally {
dispose();
runner = null;
}
}
这个 decoratedRun 就相当于上面的 SubscribeTask。run 方法中是用经过若干层封装的观察者订阅最初的那个被观察者。
SubscribeOnObserver 收到 onNext
public void onNext(T t) {
actual.onNext(t);
}
这样就进了 ObserveOnObserver 的 onNext
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
把数据先加到上面创建的那个队列里。
void schedule() {
// 如果之前是 0
if (getAndIncrement() == 0) {
// 这里就是切换接收数据的线程
worker.schedule(this);
}
}
线程执行自己的 run 方法
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
void drainNormal() {
...
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (;;) {
...
for (;;) {
...
try {
v = q.poll(); // 取出队列里的值
} catch (Throwable ex) {
...
}
...
a.onNext(v); // 发射给观察者
}
...
}
}
// 这个暂时不清楚是何时调用
void drainFused() {
...
for (;;) {
...
actual.onNext(null); // 发射空值
if (d) {
ex = error;
if (ex != null) {
actual.onError(ex); // 发射事件
} else {
actual.onComplete();
}
worker.dispose();
return;
}
...
}
}
最后将数据发射给最外层的观察者,即这个 actual,同时运行在 observerOn 指定的线程上。
mermaid Observable
mermaid Observer
plantuml