Scheduler
通过 Scheduler 来控制被观察者在哪个线程发射，观察者在哪个线程接收。默认情况，发射时在哪个线程，接收就在哪个线程。
RxJava 内置了几个 Scheduler，通过 Schedulers 来获取。
- Schedulers.trampoline()：当其它排队的任务完成后，在当前线程排队开始执行，FIFO。
- Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
- Schedulers.single()：拥有一个线程，和 newThread 相比，这个线程可以共用。
- Schedulers.computation()：计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
- Schedulers.io()：I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
切换线程
- subscribeOn(): 控制事件产生的线程。
- observeOn(): 控制事件消费的线程。
- unsubscribeOn(): 控制解除订阅时的线程。
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
源码分析
对象创建
- Observable 通过 create 方法创建 ObservableCreate，将参数 ObservableOnSubscribe 作为 ObservableCreate 的 source。
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
...
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
- 在上面返回的 ObservableCreate 上调用 subscribeOn，创建 ObservableSubscribeOn，ObservableCreate 对象作为 source。
public final Observable<T> subscribeOn(Scheduler scheduler) {
...
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
- 在上面返回的 ObservableSubscribeOn 上调用 observeOn，创建 ObservableObserveOn，ObservableSubscribeOn 自己作为 ObservableObserveOn 的 source。
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)));
}
-
第一步对下级的被观察者抛上来的观察者再封装一次，这个 s 其实就是 ObservableObserveOn 中封装后的 ObserveOnObserver。
-
第二步调用 onSubscribe，看 ObservableObserveOn 的 onSubscribe 方法
public void onSubscribe(Disposable s) { if (DisposableHelper.validate(this.s, s)) { this.s = s; // 创建一个队列 queue = new SpscLinkedArrayQueue<T>(bufferSize); // actual 是下级的观察者，这里最外层用户传进来的那个 Observer 就收到了 onSubscribe 回调 actual.onSubscribe(this); } } -
第三步是最主要的开始发射数据
向下发射数据
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
网友评论