java - 链式 API 调用中的 RxJava 错误处理

Question

我最近在尝试使用 Rxjava 来实现一系列事件(Api callas/数据库操作)，但在处理错误时似乎遇到了障碍。

这就是我想要做的。我正在调用一个 Api，它将检查用户是否存在于数据库中。根据我得到的响应，我正在尝试使用 rxjava 链接一些序列。下图可能会更好地解释。

                          checkUser()
                         /          \
                       No           Yes
                       /              \
            createUserRemote()       FetchUserNotesRemote()
                      |                    |
                    End               SaveUserNotesLocal()
                                            |
                                           End

我可以使用以下代码将 checkUser() -> FetchUserNotesRemote() -> SaveUserNotesLocal() 序列链接在一​​起。

checkUser()
            .flatMap(id -> {return fetchData(id);})
            .flatMap(notesResponseObject -> {return saveFetchedData(notesResponseObject);})
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new SingleObserver<Integer>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onSuccess(Integer integer) {
                    //handle onsuccess here
                }

                @Override
                public void onError(Throwable e) {
                    //handle errors here
                }
            });

我主要想解决的问题。

• 我不知道如何处理 checkUser() 返回的情况
  404 http 状态。因为当这种情况发生时，订阅者的 onError
  方法被调用，在我看来这是应该发生的事情。我怎么能够 处理它，以便当我从 API 收到错误 (404) 响应时， 而不是执行 FetchUserNotesRemote() 和 SaveUserNotesLocal()， 我执行不同的事件链？
• 我不确定的另一件事是，如果调用时出现错误 链中的任何可观察对象，订阅者的 onError 方法如何知道 哪个 observable 调用它？

Answer 1

1) 要在出错时执行不同的可观察链，您可以使用方法 onErorrResumeNext()。更多信息在这里:github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators

例子:

checkUser().flatMap(id -> {return fetchData(id);})
           .flatMap(notesResponseObject -> {return saveFetchedData(notesResponseObject);})
           .onErrorResumeNext(throwable -> { return doSomethingDifferent(); }
           .subscribeOn(Schedulers.io())
           .observeOn(AndroidSchedulers.mainThread())
           .subscribe(new SingleObserver<Integer>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onSuccess(Integer integer) {
                    //handle onsuccess here
                }

                @Override
                public void onError(Throwable e) {
                    //handle errors here
                }
            });

2) 如果在您的流中某处抛出异常，它会传递给订阅者 onError()。如果你想知道在流的哪一部分抛出错误，你可以添加多个 onErorrResumeNext() 调用，在每次 api 调用后抛出具体的异常。

    checkUser()
           .onErrorResumeNext(throwable -> { return Observable.error(new CheckUserException()); }
           .flatMap(id -> {return fetchData(id);})
           .onErrorResumeNext(throwable -> { return Observable.error(new FetchDataException()); }
           .flatMap(notesResponseObject -> {return saveFetchedData(notesResponseObject);})
           .onErrorResumeNext(throwable -> { return Observable.error(new SaveDataException()); }
           .subscribeOn(Schedulers.io())
           .observeOn(AndroidSchedulers.mainThread())
           .subscribe(new SingleObserver<Integer>() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onSuccess(Integer integer) {
                    //handle onsuccess here
                }

                @Override
                public void onError(Throwable e) {
                    //handle errors here
                }
            });