使用 RxJava 链接 Retrofit 调用

Posted 2023-04-18

【中文标题】使用 RxJava 链接 Retrofit 调用

【英文标题】：Chaining Retrofit calls using RxJava

【发布时间】：2017-04-22 10:56:09

【问题描述】：

我正在尝试让自己进入 RxJava，所以我阅读了一些关于它的帖子。我想我理解它是如何工作的，但我想向您提交一个理论代码，以确保我对图书馆有很好的理解。

假设我有一个 API，可以检索制作人列表，并为每个人检索他们制作的艺术家、他们的专辑和歌曲。 我的模型将是以下

public class Produceur 
    private Integer id;
    private String name;
    private String pictureUrl;


public class Artist 
    private Integer id;
    private String name;
    private String lastname;
    private String sceneName;
    private String pictureUrl;


public class Album 
    private Integer id;
    private int year;
    private String name;
    private String style;
    private String pictureUrl;
    private Integer artistId;


public class Song 
    private Integer id;
    private String title;
    private int duration;
    private String pictureUrl;
    private Integer albumId;

与我的再生服务

@GET("myUrl/produceurs")
Observable<List<ProduceurResponse>> getProduceurs();

@GET("myUrl/produceurs/produceurId")
Observable<List<ArtisteResponse>> getArtistForProduceur(@Path("produceurId") Integer produceurId);

@GET("myUrl/picture/id")
Observable<ResponseBody> getPicture(@Path("id") Integer id);

和响应对象

public class ProduceurResponse 
    private Integer id;
    private String name;
    private String pictureUrl;


public class ArtisteResponse 
    private Integer id;
    private String name;
    private String lastname;
    private String sceneName;
    private String pictureUrl;
    private List<AlbumResponse> albums;


public class AlbumResponse 
    private Integer id;
    private int year;
    private String name;
    private String style;
    private String pictureUrl;
    private Integer artistId;
    private List<SongResponse> songs;


public class SongResponse 
    private Integer id;
    private String title;
    private int duration;
    private String pictureUrl;
    private Integer albumId;

我想检索所有生产者和每个艺术家，并将所有图像以相同的顺序保存在本地内存中。 我想到了下面的代码，我们将为每个制作人检索艺术家、他们的专辑和歌曲并将它们插入到我们的基础中。 一旦我们完成对制作人和艺术家的检索，我们就可以下载图片（我们将每个 id 和图片存储在一个列表中）。

getProduceurs().flatMap(produceurs -> Observable.from(produceurs))
               .doOnNext(produceur -> insertProduceurInBase(produceur))
               .subscribe(produceur -> Observable.from(getArtistForProduceur(produceur.getId())
                                                 .flatMap(artists -> Observable.from(artists))
                                                 .doOnNext(artist -> insertArtistInBase(artist)))
                                                 .subscribe(),

                          e -> showError(e),

                          () -> Observable.from(listOfIdsToDownload)
                                          .doOnNext(id -> getPicture(id))
                                          .subscribe(response -> createImage(response),

                                          e -> showError(e),

                                          () -> isFinished()
                                          )
                         );

此代码是否有效（我认为可以，但我不确定）？这是最好的方法吗？

现在，如果我的 getProduceurs 服务向我返回一个包含制作人艺术家 ID 的 ProduceurResponse 列表，并且我获得了一项服务来检索艺术家个人资料，而另一个服务则检索其专辑。

public class ProduceurResponse 
    private Integer id;
    private String name;
    private String pictureUrl;
    List<Integer> artistsIds;


public class ArtisteProfileResponse 
    private Integer id;
    private String name;
    private String lastname;
    private String sceneName;
    private String pictureUrl;


@GET("myUrl/artist/artistId")
Observable<List<ArtisteProfileResponse>> getArtistProfile(@Path("artistId") Integer artistId);

@GET("myUrl/artist/artistId/detail")
Observable<List<AlbumResponse>> getArtistAlbums(@Path("artistId") Integer artistId);

我可以使用 .zip 使 getArtistProfile() 和 getArtistAlbums() 调用同时调用类似

getProduceurs().flatMap(produceurs -> Observable.from(produceurs))
               .doOnNext(produceur -> insertProduceurInBase(produceur))
               .subscribe(produceur -> Observable.from(produceur.getArtistIds())
                                                 .zip(
                                                        getArtistProfile(),
                                                        getArtistAlbums(),
                                                        (artistProfil, albumList) -> insertArtistInBase()
                                                 )
                                                 .subscribe(),

                          e -> showError(e),

                          () -> Observable.from(listOfIdsToDownload)
                                          .doOnNext(id -> getPicture(id))
                                          .subscribe(response -> createImage(response),

                                          e -> showError(e),

                                          () -> isFinished()
                                          )
                         );

但我真的不确定我是否以正确的方式使用 zip。 zip 在这段代码中使用得很好吗？这行得通吗？这是最好的方法吗？

编辑

所以我尝试使用 google book api 实现类似于我最初想法的东西。

我有一个 Retrofit 界面

public interface IBookService 
    @GET("volumes?q=robot+subject:fiction")
    Observable<BookSearchResult> getFictionAuthors(@Query("category") String key);
    @GET("volumes")
    Observable<BookSearchResult> getBooksForAuthor(@Query("q") String author, @Query("category") String key);

与

public class BookSearchResult 
    public List<BookResult>  items;


public class BookResult 
    public String id;
    public String selfLink;
    public VolumeInfoResult volumeInfo;
    public SaleInfoResult saleInfo;

我尝试使用字符串机器人 (getFictionAuthors) 检索小说书籍，并返回包含 BookResult 列表的 BookSearchResult。对于每本书的结果，我使用 getBooksForAuthor 检索作者的所有书籍。我的代码如下

Observable<BookResult> observable = mWebService.getFictionAuthors(API_KEY)
                .flatMap(new Func1<BookSearchResult, Observable<BookResult>>() 

                    // Parse the result and build a CurrentWeather object.
                    @Override
                    public Observable<BookResult> call(final BookSearchResult data) 
                        return Observable.from(data.items);
                    
                )
                .concatMap(new Func1<BookResult, Observable<BookSearchResult>>() 

                    // Parse the result and build a CurrentWeather object.
                    @Override
                    public Observable<BookSearchResult> call(final BookResult data) 
                        return mWebService.getBooksForAuthor("=inauthor:" + data.volumeInfo.authors.get(0), API_KEY);
                    
                )
                .flatMapIterable(new Func1<BookSearchResult, List<BookResult>>() 

                    // Parse the result and build a CurrentWeather object.
                    @Override
                    public List<BookResult> call(final BookSearchResult data) 
                        return data.items;
                    
                )
                .subscribeOn(Schedulers.newThread())
                .observeOn(androidSchedulers.mainThread())
                .subscribe(new Subscriber<BookResult>() 
                    @Override
                    public void onNext(final BookResult book) 
                        Log.e("Book","Book is " + book.volumeInfo.title + " written by " + book.volumeInfo.authors.get(0));
                    

                    @Override
                    public void onCompleted() 
                        Log.e("Book","Book list completed");
                    

                    @Override
                    public void onError(final Throwable error) 
                        Log.e("Book","Book list error");
                    
                

此代码正在运行，但有一些我不明白的奇怪之处。在我的日志中，我首先收到第一作者的 getBooksForAuthor 请求的返回，然后是该作者每本书的日志。之后，我得到了第二作者的请求结果和他书中的部分日志。按照其他作者请求的结果，然后是第二作者的书单和所有其他作者的书单的结尾。

为了说明这一点，我的日志看起来像

- > Return from request for Author 1
- > Book 1 from author 1
...
- > Book 10 from author 1
- > Return from request for Author 2
- > Book 1 from author 2
...
- > Book 5 from author 2
- > Return from request for Author 3
- > Return from request for Author 4
...
- > Return from request for Author 10
- > Book 6 from author 2
..
- > Book 10 from author 2
- > Book 1 from author 3
..
- > Book 10 from author 3
...
- > Book 1 from author 10
..
- > Book 10 from author 10

当我预料到的时候

- > Return from request for Author 1
- > Book 1 from author 1
...
- > Book 10 from author 1
- > Return from request for Author 2
- > Book 1 from author 2
...
- > Book 10 from author 2
...
- > Return from request for Author 10
- > Book 1 from author 10
...
- > Book 10 from author 10

有人解释或理解我错过了什么吗？

【问题讨论】：

您到底想要什么，想按顺序保存图像对吗？

我想检索我的所有数据并以相同的顺序保存我的图像。我想到的是我登录，我的应用程序检索我需要的所有数据，然后我可以使用该应用程序。

rxjava 有这个操作符：merge() - 它按顺序获取从 api 接收的数据。 buffer() ：此运算符创建缓冲区，以便没有负载。你可以试试这个

您必须创建一个 BaseActivity 来执行所有 Api 调用，以便您可以使用它们

根据我对merge的理解，它的目的是将Observables组合成一个单独的Observable。所以它可能会取代拉链，但我不知道我还能用它做什么。我误解了合并功能吗？

【参考方案1】：

您应该避免嵌套订阅（查看 flatmap 运算符）。这是代码异味的标志。 为避免嵌套订阅，您可以使用运算符flatMap（不要对结果事件排序）或concatMap（对结果事件排序）。

我也注意到你在完成的回调中使用了另一个Observable：在这种情况下你可以concat observables。

所以使用这种运算符对代码进行返工：

 getProduceurs().flatMap(produceurs -> Observable.from(produceurs))
                .doOnNext(produceur -> insertProduceurInBase(produceur))
                // call getArtistForProduceur and emit results in order
                .concatMap(produceur -> getArtistForProduceur(produceur.getId()))
                // emits items of the list
                .flatMapIterable(artists -> artists)
                .doOnNext(artist -> insertArtistInBase(artist)))
                // don't care about elements. But will wait for the completion of the previous observable
                .ignoreElements()
                // perform jobs after the previous observable complete
                .concatWith(Observable.from(listOfIdsToDownload)
                                      .doOnNext(id -> getPicture(id))
                                      .doOnNext(response -> createImage(response)))
                // show an error if an error occur in the downstream
                .doOnError(e -> showError(e))
                // call isFinished when everything is finished.
                .doOnCompleted(() -> isFinished())
                .subscribe()

【讨论】：

非常感谢，看起来好多了，我不知道 concatMap。我还有一个问题：flatMapIterable(artists -> Artists) 和 flatMap(artists -> Observable.from(artists)) 一样吗？

是的。见：reactivex.io/RxJava/javadoc/rx/…

感谢您的帮助！

我实现了类似于您回答的内容，但我不完全理解发生了什么。当我使用 ignoreElements() 然后使用 flatMap 或 concatMap 时，什么也没有发生（我的意思是，我的 concatMap 或 flatMap 代码没有被执行），但是当使用 concatWith 时它正在工作。我不明白为什么。你能解释一下吗？

ignoreElements ... 忽略元素。所以你不会收到新元素的通知。这就是为什么 flatMap / concatMap 永远不会收到新元素的通知。 （不要将 concatMap 与 concatWith 混淆。）