RxJava的.debounce()干扰了我的Observable的线程和错误处理

Question

我想在我的Android应用程序中进行标准搜索,我在其中输入EditText,等待用户完成输入,然后使用Retrofit启动网络请求:

// make observable out of EditText
Observable<OnTextChangeEvent> textObs = WidgetObservable.text(searchText);

mSearchResultSubscription =
    textObs

        // wait until user has not typed for 350 milliseconds
        .debounce(350, TimeUnit.MILLISECONDS)

        // get the string the user typed
        .map(OnTextChangeEvent::text)
        .map(CharSequence::toString)

        // start a new observable (from Retrofit)
        .flatMap(
            q ->
                // try network call and return my data
                MyRetrofitAPI.getService().search(q)

                    // if this fails, just return empty observable
                    .onErrorResumeNext(error -> {
                        Log.e("Error from retrofit: " + error.getLocalizedMessage());
                        return Observable.empty();
                    })

        )

        // if all is well, show the contents on the screen somehow
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(a -> {
                mAdapter.setItems(a);
            }
            , error -> {
                Log.e("Also error in outer observable: " + error.getLocalizedMessage());
            }
        );

现在,我有一个测试服务器,它接受改进调用并返回一个列表.当我输入'crash'时,服务器执行一些无效的代码和错误,返回http状态代码500,以及一些错误html.因此,改装调用失败了.

我认为外部的Observable链不应该受此影响.请参阅我之前的问题:在RxJava中,如何在出错时重试/恢复,而不是完成observable

但是,外部Observable也有错误,导致链终止.错误是: The current thread must have a looper!

奇怪.现在我尝试没有.debounce()和相同的事情发生,服务器有一个内部错误,但外部Observable 没有错误.

那么它对.debounce()导致这种行为的线程有什么影响呢？我该如何解决它？

Answer 1

添加到pturner的答案,Scheduler可以作为传递

.debounce(400, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())

这将导致观察者在Android的主线程(UI线程)上工作,并将阻止Observer抛出错误.

Answer 2

似乎问题可能在于 debounce 创建一个新线程来执行代码，来自文档：

此变体默认在计算调度程序上运行，但您可以选择传入您选择的调度程序作为第三个参数。

您可能需要传入一个带有后台 Looper 的 Android 调度程序，这将解决该错误（理论上，不幸的是，如果没有服务器，现在无法运行您的东西）。