RxJava2使用Compose后切换线程订阅触发IllegalStateException问题咨询

老哥，你遇到的这个IllegalStateException: Expected to be called on the main thread其实很常见，我来给你拆解原因和解决方案：

为啥会报错？ #

大概率是你的ObservableTransformer内部藏着必须在主线程执行的代码（比如Android的UI操作、某些SDK的主线程要求调用），但你在外部用subscribeOn(Schedulers.io())把整个上游逻辑（包括Transformer里的代码）都扔去IO线程跑了，这就触发了主线程检查的异常。当然也有可能是线程切换的顺序搞混了，但核心问题还是代码跑在了不该跑的线程上。

正确实现「计算在IO，回调在主线程」的两种方式 #

方式1：把线程逻辑整合到Transformer里（推荐，更封装） #

把线程切换直接写在Transformer内部，这样外部调用的时候不用再操心线程，逻辑更清晰：

ObservableTransformer<Event, Model> event() {
    return upstream -> upstream
            // 指定Transformer里的计算逻辑在IO线程执行
            .subscribeOn(Schedulers.io())
            .map(event -> {
                // 这里的计算逻辑安全跑在IO线程，放心搞耗时操作
                return convertEventToModel(event);
            })
            // 直接把结果切换到主线程，下游订阅时不用再处理线程
            .observeOn(AndroidSchedulers.mainThread());
}

// 调用的时候直接用，省心
Observable.just(new Event())
        .compose(event())
        .subscribe(model -> {
            // 这里的回调绝对在主线程，放心更UI
            updateUI(model);
        }, throwable -> {
            // 错误回调也在主线程，弹窗啥的随便搞
            showErrorToast(throwable);
        });

方式2：外部添加线程切换（适合Transformer是纯计算逻辑的场景） #

如果你的Transformer只是纯计算，没有任何主线程依赖的代码，那可以在外部灵活添加线程切换，注意顺序就行（其实subscribeOn的位置不影响，它只会生效一次，但observeOn要放在subscribe之前）：

// 你的Transformer：纯计算，无主线程依赖
ObservableTransformer<Event, Model> event() {
    return upstream -> upstream
            .map(event -> {
                // 纯计算，随便哪个线程都能跑
                return convertEventToModel(event);
            });
}

// 调用时手动加线程切换
Observable.just(new Event())
        .subscribeOn(Schedulers.io()) // 上游（包括Transformer）在IO线程执行
        .compose(event())
        .observeOn(AndroidSchedulers.mainThread()) // 下游订阅回调切到主线程
        .subscribe(model -> {
            updateUI(model);
        });

几个关键知识点要记牢 #

• subscribeOn 只生效一次：不管你调用多少次，只有第一次的subscribeOn会决定整个上游的执行线程（因为RxJava的订阅是从下游往上游走的，第一个碰到的subscribeOn会被用）。
• observeOn 可以多次用：每调用一次，就会把后续的操作切换到指定线程，比如你可以先切IO做计算，再切主线程更UI，再切IO做后续处理，完全灵活。
• 如果Transformer里必须有主线程操作：比如某些SDK要求必须在主线程初始化，那绝对不能把subscribeOn设为IO，这时候要拆分逻辑，把主线程操作单独放在observeOn(AndroidSchedulers.mainThread())之后：

ObservableTransformer<Event, Model> event() {
    return upstream -> upstream
            // 先切主线程做必须的操作
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext(event -> {
                // 这里的代码安全跑在主线程
                checkAppState(event);
            })
            // 切回IO做计算
            .observeOn(Schedulers.io())
            .map(event -> {
                return convertEventToModel(event);
            })
            // 最后切回主线程给下游
            .observeOn(AndroidSchedulers.mainThread());
}

这样就能完美避开线程异常，同时满足你的需求啦！

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