地图函数中的Android RxJava异步调用答案

【问题标题】：Android RxJava asynchronous call in map function地图函数中的Android RxJava异步调用
【发布时间】：2017-01-19 22:14:32
【问题描述】：

在更改“SortBy”时，我的程序将执行 NetworkIO 来检索热门电影并显示它们。

然而，似乎虽然我已经完成了subscribeOn(Schedulers.io())，但map 中的函数call 中的NetworkIO MovieDB.getPopular() 和MovieDB.getTopRated() 在主线程上执行，我得到了一个android.os.NetworkOnMainThreadException。

我想知道如何使 public Movie[] call(SortBy sortBy) 异步。

sortObservable.map(new Func1<SortBy, Movie[]>() {
    @Override
    public Movie[] call(SortBy sortBy) {
        try {
            switch (sortBy) {
                case POPULAR:
                    return MovieDB.getPopular(); // NETWORK IO
                case TOP_RATED:
                    return MovieDB.getTopRated(); // NETWORK IO
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (JSONException e) {
            e.printStackTrace();
        }
        return new Movie[0];
    }
})
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Movie[]>() {
            @Override
            public void call(Movie[] movies) {
                imageAdapter.loadData(movies);
            }
        });

【问题讨论】：

您可以改用flatMap，并将您的 io 调用包装在一个 observable 中。
subscribeOn 是调用subscribe 的位置。使用observeOn控制上一个observable的线程。
@njzk2 我还是得到了android.os.NetworkOnMainThreadException，源代码在这里：github.com/zizhengwu/Popular-Movies-Stage-1/blob/load-image/app/…
@kmx 不，这不是您将其包装在可观察对象中的方式。这不会改变 io 调用是在调用 flatmap 方法的同一线程中进行的事实。请改用可调用对象和fromCallable，这样您就可以在订阅时给它一个线程来使用
嗨@njzk2，我仍然不知道如何将callable 和fromCallable 与我的地图功能联系起来。你能给我一些代码吗？

标签： java android rx-java rx-android

【解决方案1】：

请检查以下内容是否适合您。它使用flatMap 而不是map。

sortObservable.flatMap(new Func1<SortBy, Observable<Movie[]>>() {

        @Override
        public Observable<Movie[]> call(SortBy sortBy) {
            try {
                switch (sortBy) {
                    case POPULAR:
                        return Observable.just(MovieDB.getPopular()); // NETWORK IO
                    case TOP_RATED:
                        return Observable.just(MovieDB.getTopRated()); // NETWORK IO
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (JSONException e) {
                e.printStackTrace();
            }
            return Observable.just(new Movie[0]);
        }
    }).subscribe(new Action1<Movie[]>() {
        @Override
        public void call(Movie[] movies) {
            imageAdapter.loadData(movies);
        }
    });

从您在 Github 上的源代码来看，您似乎正在使用 OkHttp 执行请求的同步模式。 OkHttp 还支持异步请求，这是首选。以下是一些方法所需的更改。

run 方法应该使用 enqueue 而不是 execute。

Observable<String> runAsync(String url){
return Observable.create(subscriber -> {
    Request request = new Request.Builder().url(url).build();

    client.newCall(request).enqueue(new Callback() {

        @Override
        public void onResponse(Call call, Response response) throws IOException {
            subscriber.onNext(response.body().string());
        }

        @Override
        public void onFailure(Call call, IOException e) {
            subscriber.onError(e);
        }
    });
});
}

getApi 可以返回 Observable<Movie[]> 而不是 Movie[]

public Observable<Movie[]> getApiAsync(String type){
return runAsync("http://api.themoviedb.org/3/movie/" + type
        + "?api_key=412e9780d02673b7599233b1636a0f0e").flatMap(response -> {
            Gson gson = new Gson();
            Map<String, Object> map = gson.fromJson(response,
                    new TypeToken<Map<String, Object>>() {
                    }.getType());
            Movie[] movies = gson.fromJson(gson.toJson(map.get("results")),
                    Movie[].class);
            return Observable.just(movies);
        });
}

【讨论】：

仍然得到android.os.NetworkOnMainThreadException。源代码在这里：github.com/zizhengwu/Popular-Movies-Stage-1/blob/load-image/app/…
该问题与 rx 无关，您可以使用 AsyncTask 来解决这个问题。您可以参考：stackoverflow.com/a/6343299/3940047
我只是好奇为什么我的代码不起作用。据我了解，网络 IO 应该在不同的线程中。我之前关注过这个stackoverflow.com/questions/32687921/using-rxjava-and-okhttp，它对我来说效果很好。
虽然不确定那部分，但已经对在异步模式下使用 OkHttp 的答案进行了更改。请看一看。

【解决方案2】：

最后我自己整理了一下：

sortObservable.flatMap(new Func1<SortBy, Observable<Movie[]>>() {

    @Override
    public Observable<Movie[]> call(SortBy sortBy) {
        switch (sortBy) {
            case POPULAR:
                return Observable.fromCallable(() -> MovieDB.getPopular()).subscribeOn(Schedulers.io());
            case TOP_RATED:
                return Observable.fromCallable(() -> MovieDB.getTopRated()).subscribeOn(Schedulers.io());
            default:
                return Observable.fromCallable(() -> new Movie[0]).subscribeOn(Schedulers.io());
        }
    }
})
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Movie[]>() {
            @Override
            public void call(Movie[] movies) {
                imageAdapter.loadData(movies);
            }
        });

【讨论】：