带有 Retrofit 2 和 RxJava 的 concatMap() 中的 NetworkOnMainThreadException答案

【问题标题】：NetworkOnMainThreadException in concatMap() with Retrofit 2 and RxJava带有 Retrofit 2 和 RxJava 的 concatMap() 中的 NetworkOnMainThreadException
【发布时间】：2016-09-02 13:19:08
【问题描述】：

在我的应用程序中，当用户搜索新电影并从网络服务检索电影数据时，我尝试使用 RxJava 做出反应。为了做到这一点，我想使用 concatMap 将获取新电影查询的动作连接到向 web 服务请求电影。这样做我得到一个 NetworkOnMainThread 异常，我不明白原因..

 createSearchViewObservable(searchView)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .concatMap(new Func1<String, Observable<MoviesWrapper>>() {
                @Override
                public Observable<MoviesWrapper> call(String title) {
                    RestMovieSource repo = new RestMovieSource();
                   return repo.searchMovieByTitle(title);
                }
            })
            .subscribe(new Subscriber<MoviesWrapper>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                    Toast.makeText(getContext(), e.getCause().getMessage(), Toast.LENGTH_SHORT).show();
                }

                @Override
                public void onNext(MoviesWrapper moviesWrapper) {
                    for (Movie movie : moviesWrapper.getResults()) {
                        Toast.makeText(getContext(), movie.getTitle(), Toast.LENGTH_SHORT).show();
                    }
                }
            });

D/OkHttp: --> GET /3/search/movie?api_key=xxxx&query=Jurassic%20World HTTP/1.1
D/OkHttp: --> END GET

  android.os.NetworkOnMainThreadException
      at android.os.StrictMode$AndroidBlockGuardPolicy.onNetwork(StrictMode.java:1147)
      at java.net.InetAddress.lookupHostByName(InetAddress.java:418)
      at java.net.InetAddress.getAllByNameImpl(InetAddress.java:252)
      at java.net.InetAddress.getAllByName(InetAddress.java:215)
      at com.squareup.okhttp.Dns$1.lookup(Dns.java:39)
      at com.squareup.okhttp.internal.http.RouteSelector.resetNextInetSocketAddress(RouteSelector.java:184)
      at com.squareup.okhttp.internal.http.RouteSelector.nextProxy(RouteSelector.java:153)
      at com.squareup.okhttp.internal.http.RouteSelector.next(RouteSelector.java:95)
      at com.squareup.okhttp.internal.http.HttpEngine.createNextConnection(HttpEngine.java:345)
      at com.squareup.okhttp.internal.http.HttpEngine.connect(HttpEngine.java:328)
      at com.squareup.okhttp.internal.http.HttpEngine.sendRequest(HttpEngine.java:246)
      at com.squareup.okhttp.Call.getResponse(Call.java:276)
      at com.squareup.okhttp.Call$ApplicationInterceptorChain.proceed(Call.java:234)
      at com.squareup.okhttp.logging.HttpLoggingInterceptor.intercept(HttpLoggingInterceptor.java:180)
      at com.squareup.okhttp.Call$ApplicationInterceptorChain.proceed(Call.java:223)
      at com.squareup.okhttp.Call.getResponseWithInterceptorChain(Call.java:196)
      at com.squareup.okhttp.Call.execute(Call.java:79)
      at retrofit.OkHttpCall.execute(OkHttpCall.java:116)
      at retrofit.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:111)
      at retrofit.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:88)
      at rx.Observable$2.call(Observable.java:162)
      at rx.Observable$2.call(Observable.java:154)
      at rx.Observable$2.call(Observable.java:162)
      at rx.Observable$2.call(Observable.java:154)
      at rx.Observable.unsafeSubscribe(Observable.java:8171)
      at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:172)
      at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:136)
      at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:79)
      at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
      at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.pollQueue(OperatorObserveOn.java:202)
      at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber$2.call(OperatorObserveOn.java:162)
      at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
      at android.os.Handler.handleCallback(Handler.java:739)
      at android.os.Handler.dispatchMessage(Handler.java:95)
      at android.os.Looper.loop(Looper.java:135)
      at android.app.ActivityThread.main(ActivityThread.java:5221)
      at java.lang.reflect.Method.invoke(Native Method)
      at java.lang.reflect.Method.invoke(Method.java:372)
      at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:899)
      at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:694)

【问题讨论】：

我刚刚意识到在 concatMap() 中使用 return repo.searchMovieByTitle(title).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()) 使代码工作。因为我没有找到一个很好的例子，所以这是最好的方法吗？

标签： android retrofit rx-java

【解决方案1】：

我解决了为从 concatMap() 返回的 observable 定义 subscribeOn 和 observeOn 的问题。外部的 subscribeOn/observeOn 仅特定于 createSearchViewObservable(searchView) 中的第一个 observable。

 createSearchViewObservable(searchView)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .concatMap(new Func1<String, Observable<MoviesWrapper>>() {
            @Override
            public Observable<MoviesWrapper> call(String title) {
                RestMovieSource repo = new RestMovieSource();
               return repo.searchMovieByTitle(title)
                      .subscribeOn(Schedulers.io())  
                      .observeOn(AndroidSchedulers.mainThread());;
            }
        })
        .subscribe(...)

【讨论】：

是的，这行得通。有没有人能够为此更改提供确切的更改日志？它曾经与 Retrofit 1 一起工作得很好......

【解决方案2】：

只需在concatMap() 之后调用subscribeOn()：

createSearchViewObservable(searchView)
        .concatMap(new Func1<String, Observable<MoviesWrapper>>() {
            @Override
            public Observable<MoviesWrapper> call(String title) {
                RestMovieSource repo = new RestMovieSource();
                return repo.searchMovieByTitle(title);
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Subscriber<MoviesWrapper>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
                Toast.makeText(getContext(), e.getCause().getMessage(), Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onNext(MoviesWrapper moviesWrapper) {
                for (Movie movie : moviesWrapper.getResults()) {
                    Toast.makeText(getContext(), movie.getTitle(), Toast.LENGTH_SHORT).show();
                }
            }
        });

【讨论】：

抱歉，这不起作用！ java.lang.IllegalStateException：在 Scheduler.Worker 线程上抛出致命异常。 -> 原因：rx.exceptions.OnErrorFailedException：尝试将错误传播到 Observer.onError 时发生错误 -> 原因：android.os.NetworkOnMainThreadException
也许您需要使用flatMap() 而不是concatMap()？
flatMap() 和 concatMap() 的区别只是最后的合并算法。我建议您阅读这篇文章以了解它fernandocejas.com/2015/01/11/…
我知道，但我认为它有帮助。
它没有！问题出在执行订阅者的线程上，而 flatMap 并没有改变它。我用我找到的解决方案添加了答案。我不知道它是否是最好的，但它确实有效。

【解决方案3】：

ConcatMap 或 flatMap 将生成一个新的 observable，它从旧的 observable 中获取数据。因此，您将调度程序设置为旧的 observable，而不是新的。

【讨论】：