如何使用 RxJava/RxAndroid 在 OnNext 中抛出异常答案

【问题标题】：How to Throw an Exception in OnNext using RxJava/RxAndroid如何使用 RxJava/RxAndroid 在 OnNext 中抛出异常
【发布时间】：2018-01-15 18:49:36
【问题描述】：

我在我的 android 应用程序中使用 Retrofit 进行网络调用。现在，如果响应出现错误（可能是错误数据），我不希望执行 onComplete。请看代码sn-p，

restClient.getService().getProjectDetail(projectId)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableObserver<List<Project>>() {
                    @Override
                    public void onNext(List<Project> value) {
                        /*Something wrong in the data and I would like to execute onError*/
                    }

                    @Override
                    public void onError(Throwable e) {
                        handleError(e, 0, "");
                        hideProgressDialog();
                    }

                    @Override
                    public void onComplete() {
                        hideProgressDialog();
                    }
                });

提前致谢。

【问题讨论】：

标签： java android rx-java rx-android

【解决方案1】：

由于您的最终消费者可能会崩溃，直接的方法是捕获该异常并委托给onError：

.subscribeWith(new DisposableObserver<List<Project>>() {
    @Override
    public void onNext(List<Project> value) {
        try {
            // something that can crash
        } catch (Throwable ex) {
            // tell the upstream we can't accept any more data
            dispose();
            // do the error handling
            onError(ex);
        }
    }
    @Override
    public void onError(Throwable e) {
        handleError(e, 0, "");
        hideProgressDialog();
    }

    @Override
    public void onComplete() {
        hideProgressDialog();
    }
});

附带说明一下，RxJava 在处理可能失败的用户函数时在其自己的运算符中使用pretty much this：try-catch、cancel the source 和通过onError 发出信号。

【讨论】：

感谢您的回答。问题是，它去了 onError，但它也去了 onComplete。如何停止调用 onComplete。
onNext 失败时，你确定它会进入 onComplete 吗？
是的，我已经调试过了。
您的源必须是手工制作的，在发出终止事件之前会忽略取消。您可以通过if (!isDisposed()) hideProgressDialog(); 中的onComplete 来防止这种情况发生。

【解决方案2】：

您可以使用平面地图。例如：

restClient.getService().getProjectDetail(projectId)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .flatMap(new Function<Integer, Observable<List<Project>>>() {
                    @Override
                    public Observable<List<Project>> apply(List<Project> x) {
                        if(validate(x)){
                            return Observable.error(new Exception("Response is invalid"));
                        }else {
                          return   Observable.just(x);
                        }
                    }

                    public boolean validate(List<Project> x){
                        return x.size()==0;
                    }
                })
                .subscribeWith(new DisposableObserver<List<Project>>() {
                    @Override
                    public void onNext(List<Project> value) {
                        /*Something wrong in the data and I would like to execute onError*/
                    }

                    @Override
                    public void onError(Throwable e) {
                        handleError(e, 0, "");
                        hideProgressDialog();
                    }

                    @Override
                    public void onComplete() {
                        hideProgressDialog();
                    }
                });

其他示例：

Observable.just(1)
                .flatMap(new Function<Integer, Observable<Integer>>() {
                    @Override
                    public Observable<Integer> apply(Integer x) {
                        if(validate(x)){
                            return Observable.error(new Exception("Response is invalid"));
                        }else {
                          return   Observable.just(x);
                        }

                    }

                    public boolean validate(Integer x){
                        return x==1;
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }
                    @Override
                    public void onComplete() {
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e("ERROR", e.getMessage());
                    }
                    @Override
                    public void onNext(Integer integer) {
                        Log.d("flatMap", "onNext: " + integer.toString());
                    }
            });

【讨论】：

【解决方案3】：

您在质疑 rx-java 的基本行为。如果您只想调用 hideProgressDialog 一次。你应该从 onError 中删除它。序列必须在出现问题后销毁。

但如果您想在 onNext 中获取其他项目并避免 onError，您可以在您的可观察链上使用此方法：onErrorResumeNext

restClient.getService().getProjectDetail(projectId).onErrorResumeNext(/*Func1*/)

请考虑此方法将在 onNext 中发出 List<Project> 而不是 onError。并且 onComplete 不会调用

只要您使用改造，onNext 只会调用一次。所以更好的解决方案是第一个

【讨论】：