RxJava 线程调度和源码阅读

2018-06-05 本文已影响38人三流之路

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。

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