使用 RxJava2 和 Retrofit 在 android 中实现即时搜索时获取 java.io.InterruptedIOException

Question

所以我正在尝试使用 rxjava2 和改造来实现即时搜索， 该过程很简单，只要用户更改文本publish.onNext() 就会被调用（发布是PublishSubject 对象）。 我添加了过滤器和去抖动以及切换映射运算符，以便在文本长度大于阈值并且没有同时使用连续输入进行调用时从服务器进行搜索。

这是代码：

subject = PublishSubject.create();
    getCompositeDisposable().add(subject
            .filter(s -> s.length() >= 3)
            .debounce(300,
                    TimeUnit.MILLISECONDS)
            .switchMap(s -> getDataManager().getHosts(
                    getDataManager().getDeviceToken(),
                    s).observeOn(Schedulers.io()))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(hostResponses -> {
                getMvpView().hideEditLoading();
                if (hostResponses.size() != 0) {
                    if (this.hostResponses != null)
                        this.hostResponses.clear();
                    this.hostResponses = hostResponses;
                    getMvpView().setHostView(getHosts(hostResponses));
                } else {
                    getMvpView().onFieldError("No host found");
                }

            }, throwable -> {
                getMvpView().hideEditLoading();
                if (throwable instanceof HttpException) {
                   HttpException exception = (HttpException)throwable;
                    if (exception.code() == 401) {
                        getMvpView().onError(R.string.code_expired,
                                BaseUtils.TOKEN_EXPIRY_TAG);
                    }
                }

            })
    );

现在我的代码运行良好，我正在实现我所需要的 但是当我输入一个长字符串并按下退格按钮时出现错误，发生的情况是当我的 AutoCompleteTextView 的文本被清除时，会引发异常

这是异常的堆栈跟踪：

java.io.InterruptedIOException: thread interrupted  
at okio.Timeout.throwIfReached(Timeout.java:145)  
at okio.Okio$1.write(Okio.java:76)  
at okio.AsyncTimeout$1.write(AsyncTimeout.java:180)  
at okio.RealBufferedSink.flush(RealBufferedSink.java:216)  
at okhttp3.internal.http1.Http1Codec.finishRequest(Http1Codec.java:166)  
at okhttp3.internal.http.CallServerInterceptor.intercept(CallServerInterceptor.java:84) 
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)  
at com.facebook.stetho.okhttp3.StethoInterceptor.intercept(StethoInterceptor.java:56)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)  
at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:45)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)  
at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)  
at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)  
at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:125)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)  
at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)  
at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:200)  
at okhttp3.RealCall.execute(RealCall.java:77)  
at retrofit2.OkHttpCall.execute(OkHttpCall.java:180)  
at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:41)  
at io.reactivex.Observable.subscribe(Observable.java:10700)  
at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34)  
at io.reactivex.Observable.subscribe(Observable.java:10700)  
at io.reactivex.internal.operators.observable.ObservableObserveOn.subscribeActual(ObservableObserveOn.java:45)  
at io.reactivex.Observable.subscribe(Observable.java:10700)  
at io.reactivex.internal.operators.observable.ObservableSwitchMap$SwitchMapObserver.onNext(ObservableSwitchMap.java:126)  
at io.reactivex.observers.SerializedObserver.onNext(SerializedObserver.java:111)  
at io.reactivex.internal.operators.observable.ObservableDebounceTimed$DebounceTimedObserver.emit(ObservableDebounceTimed.java:140)  
at io.reactivex.internal.operators.observable.ObservableDebounceTimed$DebounceEmitter.run(ObservableDebounceTimed.java:165)  
at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59)  
at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51)  
at java.util.concurrent.FutureTask.run(FutureTask.java:237)  
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)  
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)  
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)  
at java.lang.Thread.run(Thread.java:762

Answer 1

内部observeOn(Schedulers.io()) 看起来不正确，因为在那之后您立即将元素移回主线程。应该是subscribeOn(Schedulers.io())。

同时删除 subscribeOn() 调用之前的 subscribe 调用，因为它应该没有实际效果，因为链订阅了顶部的 PublishSubject。

.switchMap(s -> getDataManager()
               .getHosts(getDataManager().getDeviceToken(), s)
//             .observeOn(Schedulers.io())
               .subscribeOn(Schedulers.io())   // <-------------------------
)
.observeOn(AndroidSchedulers.mainThread())
//.subscribeOn(Schedulers.io())   // <--------------------------------------
.subscribe(hostResponses -> {