嵌套 RxJava Observables 的正确方法？答案

【问题标题】：Correct way to nest RxJava Observables?嵌套 RxJava Observables 的正确方法？
【发布时间】：2016-01-12 13:17:15
【问题描述】：

我正在使用 RxJava 并嵌套 Observables，如下所示。我想在另一个 observable 中调用一个 observable，并让外部 observable 发出内部 observable onNext 的结果。它似乎工作，但我不确定这是正确的实现，因为我无法找到任何文件来确认。

public Observable<User> updateUser(final String id) {
    return Observable.create(new Observable.OnSubscribe<User>() {
        @Override
        public void call(final Subscriber<? super User> observer) {
                try {
                if (!observer.isUnsubscribed()) {
                    getUser(id).subscribe(new Action1<User>() {
                        @Override
                        public void call(User user) {
                            observer.onNext(user);
                                observer.onCompleted();
                            }
                        });
                    }
                } catch (Exception e) {
                    observer.onError(e);
                }
        }
    });
}

public Observable<User> getUser(final String id) {
...
}

【问题讨论】：

我认为首选的方法是链接 Observables，而不是嵌套它们......
你为什么要嵌套它，你可以直接返回/调用getUser？
我想嵌套它，因为它使调用代码更简单。不可观察的版本就像 public User updateUser(final String id) { //更新用户代码 return getUser(id); }
但是在你的例子中 getUser 和 updateUser 都返回 Observable 并且 updateUser 除了运行 getUser 并发出它的值之外什么都不做，所以updateUser(id) { return getUser(id); }
不要那样做。这通常意味着取消订阅、资源清理和背压被破坏并且真的很难推理。如果你想正确处理这些事情，你的代码将变得比链接 Observables/ 复杂得多

标签： java rx-java

【解决方案1】：

当您只进行Observable 转换时避免调用subscribe，因为您必须注意在 cmets 中@akarnokd 和@zsxwing 提到的所有问题。

我也会避免使用Observable.create，因为创建OnSubscribe 实现涉及考虑背压和随之而来的并发棘手业务。首选Observable.just,Observable.from,Observable.range,Observable.defer,Observable.using（还有更多，请查看 wiki）并为更高级的目的实现SyncOnSubscribe。

这段代码可能涵盖了您的用例：

public Observable<User> updateUser(final String id) {
    return getUser(id).doOnNext(user -> updateUser(user));
}

public void updateUser(User user) {
    //whatever you want here
}

public Observable<User> getUser(final String id) {
    ...
}

【讨论】：