在 Android 中使用 Room 和 RxJava 链接多个网络调用和数据库插入答案

【问题标题】：Chain multiple network calls and database inserts with Room and RxJava in Android在 Android 中使用 Room 和 RxJava 链接多个网络调用和数据库插入
【发布时间】：2019-12-09 16:01:14
【问题描述】：

假设我有一个包含一个按钮的 Activity 的 Android 应用。当我单击按钮时，我想向返回 JSON 响应的 rest API 发出几个请求。然后我解析对一个 java 对象的响应并用 Room 持久化它。对于 http 请求，我将 Volley 请求队列实现为单例。这些请求是异步的，并将它们的响应传递回 UI 线程。在那里我让 Room 持久化对象。

我这样发送我的 http 请求：

RestService.requestSomeData(context, objectId, new ResponseListener() {
    @Override
    public void onRestSuccess(String response) {    
        // parse response JSON
        // call the insert method
    }

    @Override
    public void onRestError(int code, String errorMessage) {
        // handle error
    }
}

由于 Room 强制您将查询分派给工作线程，因此我使用 RxJava 来处理任务。因此，例如，我的 Insert 方法返回一个包含在 Single<ArrayList<Integer>> 中的插入对象的 ID 的 ArrayList。然后我调用 Insert 方法并订阅这样的结果：

myDisposable = MyDatabase.getInstance().myDao()
    .insert(myObject)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(idList -> {
        Log.d(TAG, "IDs inserted: " + idList.toString());
    }, Throwable::printStackTrace);

但是，我想将多个请求链接到服务器，然后在所有请求都完成并且数据库插入准备好以更新 UI 时收到通知（例如，显示确认消息、禁用保存按钮）。我阅读了许多文章，但在任何地方都找不到如何执行这个看似简单的任务。基本上我想要实现的是：

// a some sort of container for all the observables I get from the database insertions
private Object aPoolOfObservables;

RestService.requestSomeData(context, objectId, new ResponseListener() {
    @Override
    public void onRestSuccess(String response) {    
        // parse response JSON

        aPoolOfObservables.add(MyDatabase.getInstance().myDao()
          .insert(myObject)
          .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread()));
    }
}

// repeat this n-times more
...

aPoolOfObservables.subscribe(new Listener() {
    @Override
    public void onComplete() {
        // update UI
    }
});

然后多次执行此请求并将响应添加到Single<>（或Maybe<>或Flowable<>）响应的集合中，并且不是订阅每个流而是订阅集合，因为我只关心所有操作完成。通过在前一个的onRestSuccess 中发出请求来链接它们似乎是一个非常糟糕的解决方案。你知道是否有允许这样做的 RxJava 机制吗？

是否有任何通用方法/设计模式来处理这种情况？当您想时，我可以想到很多情况，例如仅在执行多个请求并交付结果后才启用按钮。你如何在 RxJava 的上下文中创建和订阅这样的事件？我对响应式数据的工作不多，因此任何知识都会受到赞赏。

【问题讨论】：

试试Observable.zip ()。它可以同时处理多个请求，并在全部完成时通知

标签： java android rx-java2 android-room

【解决方案1】：

您可以将每个请求包装在 Single<Pair<ArrayList<Integer>, String>> 中，以存储每个请求的每个 JSON 响应。然后，与Single.zip(...)一起执行它们

private CompositeDisposable disposables;
private ArrayList<Single<Pair<ArrayList<Integer>, String>>> singles;

RestService.requestSomeData(context, objectId, new ResponseListener() {
    @Override
    public void onRestSuccess(String response) {    
        // parse response JSON

        // kotlin syntax
        singles.add(
            MyDatabase.getInstance().myDao().insert(myObject)
                .flatMap { ids: ArrayList<String> -> 
                    // transform single to include JSON response
                    return@flatMap Single.just(Pair(ids, response))
                }
        );
    }
}

// kotlin syntax
disposables.add(
    // execute all singles
    Single.zip(singles) {}.subscribe()
);

【讨论】：