改造JavaRx2线程中断

Question

我正在创建嵌套请求,如下所示(省略了一些错误处理):

return Single.create((SingleOnSubscribe<String>) emitter -> getPages()
    .subscribe(pages -> getPageData(emitter, pages), emitter::onError))
    .compose(applySchedulers());

    // ...

private void getPageData(SingleEmitter<String> emitter, List<Page> pages) {
    service.getPage(pages.get(0).id)
            .subscribe(emitter::onSuccess, e -> {
                pages.remove(0);
                getPageData(emitter, pages);
            });
}

我曾经有一个迭代解决方案,它产生了相同的结果.页面列表按顺序排序,应按原样处理.如果连接良好,这部分代码可以正常工作,但是如果我碰巧遇到了错误的连接java.io.InterruptedIOException: thread interrupted.解决这个问题的好方法是什么？

编辑:

堆栈跟踪:

W/System.err: java.io.InterruptedIOException: thread interrupted
W/System.err:     at okio.Timeout.throwIfReached(Timeout.java:145)
W/System.err:     at okio.Okio$2.read(Okio.java:136)
W/System.err:     at okio.AsyncTimeout$2.read(AsyncTimeout.java:237)
W/System.err:     at okio.RealBufferedSource.read(RealBufferedSource.java:46)
W/System.err:     at okhttp3.internal.http1.Http1Codec$ChunkedSource.read(Http1Codec.java:429)
W/System.err:     at okio.RealBufferedSource.read(RealBufferedSource.java:46)
W/System.err:     at okio.RealBufferedSource.exhausted(RealBufferedSource.java:56)
W/System.err:     at okio.InflaterSource.refill(InflaterSource.java:101)
W/System.err:     at okio.InflaterSource.read(InflaterSource.java:62)
W/System.err:     at okio.GzipSource.read(GzipSource.java:80)
W/System.err:     at okio.RealBufferedSource.read(RealBufferedSource.java:46)
W/System.err:     at okio.ForwardingSource.read(ForwardingSource.java:35)
W/System.err:     at retrofit2.OkHttpCall$ExceptionCatchingRequestBody$1.read(OkHttpCall.java:291)
W/System.err:     at okio.Buffer.writeAll(Buffer.java:1005)
W/System.err:     at okio.RealBufferedSource.readString(RealBufferedSource.java:190)
W/System.err:     at okhttp3.ResponseBody.string(ResponseBody.java:175)

编辑2:

getPages函数:

private Single<List<Page>> getPage() {
        return Observable.merge(service.getPage("mn").toObservable(),
                service.getPage("fc",).toObservable(),
                service.getPage("sh").toObservable())
                .map(PageParser::parseActive)
                .flatMap(Observable::fromIterable)
                .sorted((f1, f2) -> f2.wage - f1.wage)
                .toList();
}

Answer 1

也许我已经找到了解决方案：

private void getPageData(SingleEmitter<String> emitter, List<Page> pages) {
  try {
    service.getPage(pages.get(0).id)
            .subscribe(emitter::onSuccess, e -> {
                pages.remove(0);
                getPageData(emitter, pages);
            });
   } catch (InterruptedIOException e) {
         Log.d(TAG, e.getLocalizedMessage(), e);
   }
}

rxFragment当用户停止片段或活动时，您应该使用组件来中断 rxJava2 线程：

observable.compose(RxLifecycle.<NetworkResult, FragmentEvent>bindUntilEvent(lifecycle(), FragmentEvent.STOP));

但是会发生 ThreadInterrupted 异常，因为您应该自己处理 Retrofit2 的 Exception 并忽略它。这样做对我来说非常有效！