用 RxJava 替换嵌套的 Retrofit 调用

Question

首先，让我先说我对 RxJava 比较陌生，所以请耐心等待我掌握了基础知识。 :)

我基本上了解整个subscribe[On|With]() 初学者类型的东西（或多或少）。然而，我正在尝试做一些更复杂的事情，而 JavaRx 似乎非常适合这种事情。

我有一个精选的电影数据库，我从中查询电影详细信息。现在，这里的 API 有点奇怪。您必须打两个电话才能获得有关电影的所有信息。第一个 API 调用生成 id、标题、类别和评级，第二个生成其他重要内容，如描述、预告片等。您可能已经猜到，您需要第 1 部分的 movieId 来获取额外信息。嵌套调用 work 在这里，但它们绝不被认为是最佳实践。

我的代码如下所示：

pageSingle.subscribeOn(Schedulers.io())
        .subscribeWith(new DisposableSingleObserver<mPage>() {
            @Override public void onSuccess(mPage value) {
                moviesInPage = value.movies;
                if(moviesInPage==null){
                    Log.w(TAG, "mergeMovieParts: no movies returned");
                }else {
                    for (int i = 0; i < moviesInPage.size(); i++) {
                        final fMovie firstMovie = moviesInPage.get(i);
                        apiService.getSecondPart(firstMovie.id)
                                .subscribeWith(new DisposableSingleObserver<sMovie>() {
                            @Override 
                            public void onSuccess(sMovie secondMovie) {
                                mergeMovieParts(firstMovie, secondMovie);
                                }
                            @Override 
                            public void onError(Throwable e) {
                                e.printStackTrace();
                                }
                        });
                    }
                }

            }
            @Override public void onError(Throwable e) {
                handleError(e);
            }
        });

pageSingle 也是 Single。 您可能会明白为什么这不能保证一直有效。 Single.zip() 在这里不起作用，因为我需要第一次调用的信息（更具体地说是id）来启动第二次调用。我的问题归结为：

1. 如何用 JavaRx 替换那些嵌套的 Retrofit 调用 那些？请注意mPage 返回一个List<fMovie> 对象。

2. 目前，我一个接一个地进行这些调用。是吗 可以获取电影列表并基于此制作多个 同时调用以获取第二部分？我正在猜测 多线程是这里的答案，但我该怎么做 RxJava 及其含义是什么，例如值得吗，在你的 意见或权衡（如果有的话）是否很大？

3. 与这个问题无关，但你有什么建议 书籍、视频等。我可以阅读/观看这将帮助我赶上进度 使用 RxJava 最多。这让我很兴奋，但同时我 觉得通过“阅读 文档”。

如果我不够清楚，请随时要求更多上下文/澄清。提前致谢！

编辑：flatMap() 出现的错误：

Error:(109, 17) error: no suitable method found for flatMap(<anonymous Function<Movie1,ObservableSource<?>>>)
method Observable.<R#1>flatMap(Function<? super Object,? extends ObservableSource<? extends R#1>>) is not applicable
(cannot infer type-variable(s) R#1
(argument mismatch; <anonymous Function<Movie1,ObservableSource<?>>> cannot be converted to Function<? super Object,? extends ObservableSource<? extends R#1>>))
method Observable.<R#2>flatMap(Function<? super Object,? extends ObservableSource<? extends R#2>>,boolean) is not applicable
(cannot infer type-variable(s) R#2
(actual and formal argument lists differ in length))
method Observable.<R#3>flatMap(Function<? super Object,? extends ObservableSource<? extends R#3>>,boolean,int) is not applicable
(cannot infer type-variable(s) R#3
(actual and formal argument lists differ in length))
method Observable.<R#4>flatMap(Function<? super Object,? extends ObservableSource<? extends R#4>>,boolean,int,int) is not applicable
[...]

我的实现：

movieService.getMoviesFromPath(path)
                .subscribeOn(Schedulers.io())
                .flattenAsObservable(new Function<MoviePage, Iterable<?>>() {
                    @Override
                    public Iterable<?> apply(MoviePage moviePage) throws Exception {
                        return moviePage.movieList;  // movieList is a List<Movie1>
                    }
                })
                .flatMap(new Function<Movie1, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Movie1 movie1) throws Exception {
                        return null;
                    }
                });

Answer 1

您似乎需要以下内容：

movieService.getPage()
    .flattenAsObservable(page -> page.movies)
    .flatMap(movie -> apiService.getSecondPart(movie.movieId))

或者，去掉 lambda，

movieService.getPage()
            .flattenAsObservable(new Function<Page, Iterable<Movie>>() {
                @Override
                public Iterable<Movie> apply(Page page) throws Exception {
                    return page.movies;
                }
            })
            .flatMap(new Function<Movie, ObservableSource<DetailedMovie>>() {
                @Override
                public ObservableSource<DetailedMovie> apply(Movie movie) throws Exception {
                    return apiService.getSecondPart(movie.movieId);
                }
            })

话虽如此，在原帖中写new DisposableSingleObserver<mPage> 之类的东西是一个不好的迹象。在继续 Rx 之前，可能值得您花时间了解Generics in Java，因为这是您的第二个错误的原因。

您可能需要具备良好的泛型工作知识 (and standard Java naming practices) 才能充分利用 RxJava。祝你好运！

请注意，这两个示例假定 subscribeOn(Schedulers.io()) 是隐式的，并在 Retrofit 级别进行配置。你可以看看如何做到这一点in these answers

更新：如果你想将两个 Movie 对象合并在一起，我想你可以这样做：

    movieService.getPage()
            .flattenAsObservable(new Function<Page, Iterable<? extends Movie>>() {
                @Override
                public Iterable<? extends Movie> apply(Page page) throws Exception {
                    return page.movies;
                }
            })
            .flatMap(new Function<Movie, ObservableSource<DetailedMovie>>() {
                @Override
                public ObservableSource<DetailedMovie> apply(Movie movie) throws Exception {
                    return apiService.getSecondPart(movie.movieId).toObservable();
                }
            }, new BiFunction<Movie, DetailedMovie, MergedMovie>() {
                @Override
                public MergedMovie apply(Movie movie, DetailedMovie detailedMovie) throws Exception {
                    return new MergedMovie(movie, detailedMovie);
                }
            });