java - 多个 observable 完成后做某事

Question

我正在 Android 上使用 RXJava，并尝试将多个 API 调用链接在一起，并在两个 API 调用完成后执行某些操作。我的 API 调用看起来都与提供的代码示例类似。基本上进行API调用，在onNext中将每条记录写入数据库，并在所有记录都写入后，更新一些缓存。我想异步触发这两个调用，然后在两个调用都达到 onCompleted 后，再执行其他操作。 RX 中执行此操作的正确方法是什么？我认为我不需要 zip，因为我不需要将不同的流 bundle 在一起。我在想也许可以合并，但是我的两个 API 调用返回了不同类型的 Observable。请告诉我。谢谢。

    getUsers()
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .flatMap(Observable::from)
            .subscribe(new Subscriber<User>() {
                @Override
                public void onCompleted() {
                   updateUserCache();
                }

                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "Error loading users", e);
                }

                @Override
                public void onNext(User user) {
                    insertUserToDB(user);
                }
            });

    getLocations()
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .flatMap(Observable::from)
            .subscribe(new Subscriber<Location>() {
                @Override
                public void onCompleted() {
                   updateLocationCache();
                }

                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "Error loading Locations", e);
                }

                @Override
                public void onNext(Location location) {
                    insertLocationToDB(location);
                }
            });  

Answer 1

你的想法是正确的。您应该使用 zip 运算符。

您的每个函数都应该进行调用、写入数据库并执行您需要的所有操作。 Theat zip 输出功能有所不同:当调用它时，您可以确定所有 Observable 已成功完成 -> 只需完成您的 react 流。

创建一个Observable列表:

List<Observable<?>> observableList = new ArrayList<>();
observableList.add(
        getUsers()
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .flatMap(Observable::from)
            .insertUserToDB(user)
            .toList());

observableList.add(
        getLocations()
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .flatMap(Observable::from)
            .insertLocationToDB(location)
            .toList());

然后zip所有Observable的:

Observable.zip(observableList, new FuncN<Object, Observable<?>>() {
    @Override
    public Observable<?> call(Object... args) {
        return Observable.empty();
    }
}).subscribe(new Subscriber<Object>() {
    @Override
    public void onCompleted() {
        updateUserCache();
        updateLocationCache();
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Object o) {

    }
});

这是伪代码，但我希望你能理解这个想法。