RxJava2中PublishSubject与BehaviorSubject发射顺序异常问题排查

为什么会出现这个顺序问题？ #

核心原因在于**observeOn(AndroidSchedulers.mainThread())的异步调度特性**，以及事件发射和处理线程的差异：

当你调用subject1.onNext(functionA1)时，observeOn会把处理这个事件的逻辑打包成一个Runnable，通过Android的Handler.post()放到主线程消息队列的末尾——当前代码不会等待这个Runnable执行，而是直接继续往下走。

接下来你调用subject2.onNext(functionB)，如果subject2的订阅没有做线程切换（或者是在后台线程处理后再手动切回主线程），那处理functionB的逻辑会被推迟：

• 如果emittingItems是在后台线程执行，subject2的回调可能会先在后台线程运行，等它需要更新UI时再把任务post到主线程，这时候subject1的functionA1和functionA2已经排在主线程队列的前面了，最终就会出现A1 → A2 → B的顺序。
• 即使subject2也用了observeOn，如果发射线程和主线程不一致，也可能因为调度的细微差异导致顺序错乱（不过这种情况比较少见）。

另外要注意：RxJava的Subject本身是非线程安全的，如果在多线程环境下发射事件，也可能出现顺序混乱的问题。

如何实现你期望的A1 → B → A2顺序？ #

这里有几种靠谱的解决方案，你可以根据自己的场景选择：

1. 给所有Subject的订阅统一加上observeOn(AndroidSchedulers.mainThread())

确保subject2的处理逻辑也通过observeOn切换到主线程，并且让emittingItems在同一个后台线程执行：

// 给subject2的订阅加上observeOn
val disposable2 = viewModel.subject2
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(this::handleFunctionB)
disposables.add(disposable2)

这样，当你在后台线程执行emittingItems时，三个事件的处理Runnable会严格按照发射顺序被放到主线程队列，最终主线程会按A1 → B → A2的顺序处理它们。

2. 让事件发射和处理都在主线程同步执行

如果你的业务允许在主线程执行发射操作，可以把emittingItems的逻辑post到主线程，同时确保订阅的线程调度不会打乱顺序：

fun safeEmitItems() {
    Handler(Looper.getMainLooper()).post {
        subject1.onNext(functionA1)
        subject2.onNext(functionB)
        if (something) subject1.onNext(functionA2)
    }
}

这样所有事件的发射和处理都在主线程同步进行，顺序会完全符合你的预期。

3. 给Subject加上serialize()保证线程安全和顺序

Subject本身是非线程安全的，使用serialize()可以强制让事件按发射顺序被下游接收，避免多线程环境下的顺序错乱：

// 初始化Subject时加上serialize()
val subject1 = PublishSubject.create<YourFunctionType>().serialize()
val subject2 = BehaviorSubject.createDefault<YourFunctionType>(defaultValue).serialize()

这个方案能从根源上避免Subject因为多线程发射导致的顺序问题。

4. 把三个事件合并成一个Observable流

如果你希望完全掌控顺序，可以把所有事件放到一个Observable里，统一调度到主线程处理：

fun emittingItems() {
    val eventList = mutableListOf<YourFunctionType>().apply {
        add(functionA1)
        add(functionB)
        if (something) add(functionA2)
    }
    Observable.fromIterable(eventList)
        .subscribeOn(Schedulers.io()) // 可选，根据需要选择发射线程
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { event ->
            when (event) {
                is FunctionA1, is FunctionA2 -> someFunction(event)
                is FunctionB -> handleFunctionB(event)
            }
        }
}

这种方式把多个事件变成了一个有序的流，完全不用担心顺序问题。

内容的提问来源于stack exchange，提问作者MaaAn13