使用RxJava可以轻松地实现线程切换，所以在Android中常被用来替代AsyncTask、Handler等原生工具类。使用起来虽然简单，但如果不了解其背后的基本原理，很可能因为使用不当而写出bug。本文将带大家简单了解一下RxJava线程切换的实现原理以及开发中的注意事项

1. Basic Usage

• Scheduler

If you want to introduce multithreading into your cascade of Observable operators, you can do so by instructing those operators to operate on particular Schedulers.
通过Scheduler让操作符跑在指定线程，从而实现多线程调度

• observerOn

specify the Scheduler on which an observer will observe this Observable
指定一个观察者在哪个调度器上观察这个Observable

• subscribeOn

specify the Scheduler on which an Observable will operate
指定Observable自身在哪个调度器上执行

image image

RxJava调用链中每个操作符都会创建一个新的Observable，操作符产生的新Observable都会向上层的Observable注册回调。 subscribeOn和observeOn的实现原理一样：

• subscribeOn 在指定线程中向上游订阅（在指定线程中去调上游的subscribe方法）
• observeOn 收到数据后在指定线程中调用下游的回调方法（onNext/onError/onComplete等）

RxJava自下而上建立订阅，而后自上而下发射数据，所以 subscribeOn 即使出现 observeOn 之后也能保证数据源运行的线程，因为订阅永远发生在前。

2. subscribeOn

2.1 实现原理

通过源码了解一下subscribeOn实现线程切换的基本原理

//ObservableSubscribeOn.java
final class ObservableSubscribeOn extends Observable<T> {

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
        s.onSubscribe(parent);
        // 没有直接调用subscribe订阅，而是先进行了线程变换（scheduler.scheduleDirect）
        parent.setDisposable(
            scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

    final class SubscribeTask implements Runnable {
        @Override
        public void run() {
            // run()会在指定的scheduler调用，向上游订阅时线程已经发生了变化
            // 所以保证了上游所运行的线程
            source.subscribe(parent);
        }
    }

    static final
    class SubscribeOnObserver<T> implements Observer<T>, Disposable {

        @Override
        public void onNext(T t) {
            // 收到数据后不进行线程变换
            actual.onNext(t);
        }
    }
}

2.2 subscribeOn只生效一次

subscribeOn通过切换订阅线程，改变Observable.create所在线程，从而影响数据的发射线程。

由于订阅过程自下而上，所以Observable.create只受最近一次subscribeOn影响，当调用链中有多个subscribeOn时只有第一个有效。其他subscibeOn仍然可以影响其上游的doOnSubscribe的执行线程。

@Test
fun test() {
    Observable.create<Unit> { emitter ->
        log("onSubscribe")
        emitter.onNext(Unit)
        emitter.onComplete()
    }.subscribeOn(namedScheduler("1 - subscribeOn"))
        .doOnSubscribe { log("1 - doOnSubscribe") }
        .subscribeOn(namedScheduler("2 - subscribeOn"))
        .doOnSubscribe { log("2 - doOnSubscribe") }
        .doOnNext { log("onNext") }
        .test().awaitTerminalEvent() // Wait until observable completes
 }

image

2.3 正确理解subscribeOn的意义

Even though we added .subscribeOn() that is not enough. SubscribeOn operator only switches the subscribing process to the desired thread, but that doesn’t mean the items will be emitted on that thread.
subscribeOn用来决定订阅线程，但这并不意味着上游数据一定来自此线程

@Test
fun test() {
    val observable = Observable.create<Int> { emitter ->
        log("onSubscribe")
        thread(name = "Main thread", isDaemon = false) {
            log("1 - emitting"); emitter.onNext(1)
            log("2 - emitting"); emitter.onNext(2)
            log("3 - emitting"); emitter.onNext(3)
            emitter.onComplete()
        }
    }

    observable
        .subscribeOn(Schedulers.computation())
        .doOnNext { log("$it - after subscribeOn") }
        .test().awaitTerminalEvent() // Wait until observable completes
}

image

正确理解subscribeOn的含义有助于避免一些使用上的误区：

对于PublishSubject无效

@Test
fun test() {
    val subject = PublishSubject.create<Int>()
    val observer1 = subject
        .subscribeOn(Schedulers.io())
        .doOnNext { log("$it - I want this happen on an IO thread") }
        .test()
    val observer2 = subject
        .subscribeOn(Schedulers.newThread())
        .doOnNext { log("$it - I want this happen on a new thread") }
        .test()

    sleep(10); 
    subject.onNext(1)
    subject.onNext(2)
    subject.onNext(3)
    subject.onComplete()

    observer1.awaitTerminalEvent()
    observer2.awaitTerminalEvent()
}

image

对于PublishSubject来说，上游数据来自哪个线程是在onNext时决定的，所以对一个PublishSubject使用使用subscribeOn没有意义。

对于Observable.just()无效

通常subcribeOn可以决定Observable.create {...} 的执行线程，因此很多初学者容易犯的一个错误是在Observable.just(...)里做耗时任务，并误认为会跑在subscribeOn的线程:

image

如上，readFromDb() 放在just中显然是不合适的。just()在当前线程立即执行，因此不受subscribeOn影响，应该修改如下：

//Observable.defer
Observable.defer { Observable.just(readFromDb()) }
    .subscribeOn(Schedulers.io())
    .subscribe { ... }

//Observable.fromCallable
Observable.fromCallable { readFromDb() }
    .subscribeOn(Schedulers.io())
    .subscribe { ... }

image.gif

使用flatMap处理并发

subscribeOn决定的当前Observable的订阅线程，因此对于flatMap的使用要特别留心

Observable.fromIterable(listOf("id1", "id2", "id3"))
    .flatMap { id -> loadData(id) }
    .subscribeOn(Schedulers.io())
    .observeOn(mainThread())
    .toList()
    .subscribe { result -> log(result) }

如果我们希望多个loadData(id)并发执行，上述写法是错误的。

subscribeOn决定了flatMap上游线程，flatMap返回多个Observable的订阅都是发生在此线程，多个loadData只能运行在单一线程，无法实现并行。

想要达到并行执行效果，需要修改如下：

Observable.fromIterable(listOf("id1", "id2", "id3"))
    .flatMap { id ->
        loadData(id)
            .subscribeOn(Schedulers.io())
    }
    .observeOn(mainThread())
    .toList()
    .subscribe { result -> log(result) }

3.observeOn

3.1 实现原理

通过源码了解一下observeOn实现线程切换的基本原理

//ObservableObserveOn.java
final class ObservableObserveOn extends Observable<T> {

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            source.subscribe(observer);
        } else { 
            Scheduler.Worker w = scheduler.createWorker();
            // 直接向上游订阅数据，不进行线程切换，切换操作在Observer中进行
            source.subscribe(
                new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }

    static final class ObserveOnObserver<T> implements Observer<T>, Runnable {

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            // 这里选把数据放到队列中，增加吞吐量，提高性能
            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            }
            // 在schedule方法里进行线程切换并把数据循环取出
            // 回调给下游，下游会在指定的线程中收到数据
            schedule();
        }

        void schedule() {
            if (this.getAndIncrement() == 0) {
                //切换线程
                this.worker.schedule(this);
            }

        }
    }
}

3.2 observeOn多次生效

不同于subscribeOn，observeOn可以有多个而且每个都会生效

• subscribeOn切换的线程可以通过doOnSubscribe监听
• observeOn切换的线程可以通过doOnNext监听

3.3 连续发射多个item时能否保证串行？

observeOn使用Scheduler调度线程后，下游是运行在单线程中还是多个线程中？能否保证下游数据的有序性？

@Test
fun test() {
    Observable.create<Int> { emitter ->
        repeat(10) {
            emitter.onNext(it)
        }
        emitter.onComplete()
    }.observeOn(Schedulers.io())
        .subscribe {
            log(" - $it")
        }
}

image

通过结果可以看到，即使经Scheduler调度之后，下游仍然运行在单一线程，可以保证数据在整个调用链上的有序性。

那么为什么经过Scheduler调度后都跑在单一线程呢？

4. Scheduler

4.1 实现原理

image image

Scheculer并非直接调度Runnable，而是创建Worker，再由Worker来调度具体任务。

subscribeOn中的SubscribeTask以及observeOn中的ObserveOnObserver都实现了Runnable，所以最终都是在Worker中执行。

4.2 任务是由Worker调度的

image

​

image.gif

一个Scheduler可以创建多个 Worker，一个Worker可以管理多个Task（Runnable）

Worker 的存在为了确保两件事：

• 同一个 Worker 创建的 Task 确保串行执行，且立即执行的任务符合先进先出原则。
• Worker 绑定了调用了他的方法的 Runnable，当该 Worker 取消时，基于他的 Task 均被取消

4.3 Worker何保证串行？

非常简单，每个Worker只有一个线程

image image

现在可以解答疑问了：为什么observeOn经过Scheduerl调度后，仍然跑在单一线程？

Scheduler为每个observeOn分配唯一Worker，因此observeOn的下游可以保证在单一线程串行执行。

//ObservableObserveOn.java
final class ObservableObserveOn extends Observable<T> {

    @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)); //传入worker
        }
    }

  ...

}

如上，Worker作为ObserveOnObserver的成员变量被持有

4.4 预置Schedulers

如同Executors提供了多种ThreadPoolExecutor一样，Schedulers提供了多种预设的Scheduler

No.	Schedulers & Descriptions
1	Schedulers.single()
全局唯一线程，无论有多少个observables，都共享此唯一线程。
2	Schedulers.io()
最常见的调度器之一，用于IO相关操作，比如网络请求和文件操作。IO 调度器背后由线程池支撑。它首先创建一个工作线程，可以复用于其他操作，当这个工作线程(长时间任务的情况)不能被复用时，会创建一个新的线程来处理其他操作。
3	Schedulers.computation()
和IO调度器非常相似，也是基于线程池实现的。其可用的线程数是固定的，与cpu核数目保持一致。当所有线程都处于忙碌状态时，新任务只能处于等待状态。因此，它不适合IO相关操作。适用于进行一些计算操作，单一计算任务不会长时间占用线程。
4	Schedulers.newThread()
每次调用都创建新线程
5	Schedulers.trampoline()
在当前线程执行，不切换线程。
6	Schedulers.from(java.util.concurrent.Executor executor)
更像是一种自定义的IO调度器。我们可以通过制定线程池的大小来创建一个自定义的线程池。适用于observables的数量对于IO调度器太多的场景使用，

//Sample of Schedulers.from
fun namedScheduler(name: String): Scheduler {
    return Schedulers.from(
        Executors.newCachedThreadPool { Thread(it, name) }
    )
}

Thread-Safety

5.1 RxJava操作符是否线程安全？

@Test
fun test() {
    val numberOfThreads = 1000
    val publishSubject = PublishSubject.create<Int>()
    val actuallyReceived = AtomicInteger()

    publishSubject
        .take(300).subscribe {
            actuallyReceived.incrementAndGet()
        }

    val latch = CountDownLatch(numberOfThreads)
    var threads = listOf<Thread>()

    (0..numberOfThreads).forEach {
        threads += thread(start = false) {
            publishSubject.onNext(it)
            latch.countDown()
        }
    }

    threads.forEach { it.start() }
    latch.await()

    val sum = actuallyReceived.get()
    check(sum == 300) { "$sum != 300" }
}

image

结果不符合预期，因为take不是线程安全的

看一下take的源码

public final class ObservableTake<T> extends AbstractObservableWithUpstream<T, T> {
    final long limit;

    public ObservableTake(ObservableSource<T> source, long limit) {
        super(source);
        this.limit = limit;
    }
    protected void subscribeActual(Observer<? super T> observer) {
        this.source.subscribe(new ObservableTake.TakeObserver(observer, this.limit));
    }

    static final class TakeObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> downstream;
        boolean done;
        Disposable upstream;
        long remaining;

        TakeObserver(Observer<? super T> actual, long limit) {
            this.downstream = actual;
            this.remaining = limit;
        }

        public void onNext(T t) {
            if (!this.done && this.remaining-- > 0L) {
                boolean stop = this.remaining == 0L;
                this.downstream.onNext(t);
                if (stop) {
                    this.onComplete();
                }
            }

        }
    }
}

果然不出所料 remaining--没有加锁操作

5.2 observableOn的线程安全

那如果加上observableOn是不是就保证串行了呢，因为take可以跑在单一线程上了

@Test
fun test() {
    repeat(10000) {
        val numberOfThreads = 1000
        val publishSubject = PublishSubject.create<Int>()
        val actuallyReceived = AtomicInteger()

        publishSubject
            .observeOn(Schedulers.io())
            .take(300).subscribe {
                actuallyReceived.incrementAndGet()
            }

        val latch = CountDownLatch(numberOfThreads)
        var threads = listOf<Thread>()

        (0..numberOfThreads).forEach {
            threads += thread(start = false) {
                publishSubject.onNext(it)
                latch.countDown()
            }
        }

        threads.forEach { it.start() }
        latch.await()

        check(actuallyReceived.get() == 300)
    }
}

很遗憾，多次运行后发现依然有问题，因为observableOn本身也不是线程安全的，observableOn中使用的queue是一个非线程安全队列。

image ​ image

5.3 The Observable Contract

Rx在对Observable的定义中已经明确告诉我们了：

Observables must issue notifications to observers serially (not in parallel). They may issue these notifications from different threads, but there must be a formal happens-before relationship between the notifications.
reactivex.io/documentati…

作为结论，RxJava的操作符默认并非线程安全的。

但是对于接收多个Observable的操作符，例如 merge()、combineLatest()、zip()等 是线程安全的，所以即使多个Observable来自不线程时，也不需要考虑线程安全问题。