将嵌套的 Retrofit 调用替换为 RxJava

Question

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

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

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

我的代码看起来像这样：

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))

或者，拿走λ，

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 之前，可能值得您花时间了解 Java 中的泛型，因为这是导致第二个错误的原因。

您可能需要对泛型(和标准 Java 命名实践)有很好的工作知识，才能充分利用 RxJava。祝你好运！

请注意，这两个示例假定subscribeOn(Schedulers.io())是隐式的，并在改造级别进行配置。您可以在这些答案中看到如何执行此操作

更新：如果您想将两个 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);
}
});