如何处理主线程中的 RxAndroid 错误

Question

我是 rxJava/Android 的新手，很惊讶我的 onError lambda 有时会在主线程上调用，有时不会，尽管我使用 .observeOn(AndroidSchedulers.mainThread())

示例 1：onError 在主线程
这按预期工作：onError 在主线程上被调用

Observable.error(new RuntimeException("RTE"))
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(s -> {
                Log.e(TAG, "onNext("+s+")-thread: " + Thread.currentThread().getName());
            },
            throwable -> {
                Log.e(TAG, "onError()-thread: " + Thread.currentThread().getName());
            });

日志输出：

onError()-thread: main

示例 2：onError 不是在主线程上

Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("one and only");
    }
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.timeout(1, TimeUnit.SECONDS)
.subscribe(s -> {
        Log.e(TAG, "onNext("+s+")-thread: " + Thread.currentThread().getName());
    },
    throwable -> {
        Log.e(TAG, "onError()-thread: " + Thread.currentThread().getName());
    });

输出是这样的：

onNext(one and only)-thread: main
onError()-thread: RxComputationScheduler-4       

我认为调用observeOn(AndroidSchedulers.mainThread())后，所有的发射都应该在主线程上完成。

所以我有这些问题：

1. 我找不到任何说明在什么情况下onError 在哪个线程中被调用的文档。有人知道链接吗？
2. 我当然想在 GUI 中显示一些错误指示：那么如何强制 onError 在主线程中始终被调用？

Answer 1

我刚刚发现我只需要更改调用顺序。当我在timeout 之后调用observeOn 时，它按预期工作：

.timeout(1, TimeUnit.SECONDS)
.observeOn(AndroidSchedulers.mainThread())

日志输出

onNext(one and only)-thread: main
onError()-thread: main

原因是，observeOn 只会影响调用下面的所有内容，并且只会影响其他操作员再次更改线程。在上面的例子中，timeout() 将变为计算线程。

请注意，subscribeOn 的工作方式不同。您在链中的哪个位置调用它并不重要。
你应该只调用一次（当你多次调用它时，第一次调用获胜：参见Blog中的“Multiple subscribeOn”）

这是一篇不错的博文，其中包含更多详细信息：RxJava- Understanding observeOn() and subscribeOn()

Answer 2

因为.timeout(1, TimeUnit.SECONDS) 默认作用于Schedulers.computation()。