java - 嵌套 RxJava Observables 的正确方法？

Question

我正在使用 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) {
...
}

Answer 1

当您刚刚进行 Observable 转换时，请避免调用 subscribe，因为您必须注意 @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) {
    ...
}