处理 onError 和继续处理的最佳实践 [英] Best practice for handling onError and continuing processing
问题描述
我是 RxJava 的新手，但我正在将它集成到我正在处理的项目中以帮助我学习它.我遇到了一个关于最佳实践的问题.
I am new to RxJava but I am integrating it into a project that I am working on to help me learn it. I have run into a question about best practices.
我有一个关于如何处理 onError 以防止停止 Observable 处理的问题.
I have a question about how to handle onError from preventing the stopping of the Observable processing.
这里是设置:
对于我想要执行 2 个或更多网络请求的每个用户 ID，我都有一个用户 ID 列表.如果用户 ID 的任何网络请求失败，则该用户 ID 将不会更新并且可以跳过.这不应阻止其他用户标识被处理.我确实有一个解决方案，但它涉及嵌套订阅(请参阅第二个代码块).我确实看到的一个问题是，如果每次调用失败，即使在检测到某个阈值失败后，也无法短路并阻止剩余调用访问网络资源.
I have a list of userIds for each one I would like to do 2 or more network requests. If any of the network requests fails for the userid then that userid won't be updated and can be skipped. This should not prevent the other userids from being processed. I do have a solution, but it involves nested subscribes (see second block of code). One problem I do see is, if each call fails there is no way to short circuit and stop the remaining from hitting a network resource even after detecting a certain threshold number have failed.
有没有更好的方法来做到这一点?
Is there a better way to do this?
在传统代码中:
List<String> results = new ArrayList<String>();
for (String userId : userIds) {
try {
String info = getInfo(userId); // can throw an GetInfoException
String otherInfo = getOtherInfo(userId); // can throw an GetOtherInfoException
results.add(info + ", " + otherInfo);
} catch (GetInfoException e) {
log.error(e);
} catch (GetOtherInfoException e) {
log.error(e);
}
}
问题:
伪代码:
userid -> network requests -> result
1 -> a, b -> onNext(1[a ,b])
2 -> a, onError -> onError
3 -> a, b -> onNext(3[a, b])
4 -> a, b -> onNext(4[a, b])
以下是用户 ID 列表和每 2 个信息请求的工作示例.如果你运行它你会看到它会失败(见下面的源代码)
The following is a working example of a list of userIds and for each 2 requests for info. If you run it you will see that it will fail (see below the source code)
import rx.Observable;
import rx.Observable.OnSubscribeFunc;
import rx.Observer;
import rx.Subscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Action1;
import rx.util.functions.Func1;
public class TestMergeDelayError {
public static Observable<String> getUserIds() {
return Observable.from(new String[]{"1", "2", "3", "4", "5", "6"});
}
public static Observable<String> getInfo(final String prefix, final String integer, final String errorNumber) {
Observable<String> observable = Observable.create(new OnSubscribeFunc<String>() {
public Subscription onSubscribe(Observer<? super String> t1) {
if (integer.contains(errorNumber)) {
t1.onError(new Exception());
} else {
t1.onNext(prefix + integer);
t1.onCompleted();
}
return Subscriptions.empty();
}
});
return observable;
}
public static void main(String[] args) {
Observable<String> userIdObservable = getUserIds();
Observable<String> t = userIdObservable.flatMap(new Func1<String, Observable<String>>() {
public Observable<String> call(final String t1) {
Observable<String> info1 = getInfo("1::: ", t1, "2");
Observable<String> info2 = getInfo("2::: ",t1, "3");
return Observable.mergeDelayError(info1, info2);
}
});
t.subscribe(new Action1<String>() {
public void call(String t1) {
System.out.println(t1);
}
}, new Action1<Throwable>() {
public void call(Throwable t1) {
t1.printStackTrace();
}
},
new Action0(){
public void call() {
System.out.println("onComplete");
}
});
}
}
输出:
1::: 1
2::: 1
2::: 2
java.lang.Exception
at TestMergeDelayError$1.onSubscribe(TestMergeDelayError.java:32)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMergeDelayError$MergeDelayErrorObservable$ParentObserver.onNext(OperationMergeDelayError.java:266)
at rx.operators.OperationMergeDelayError$MergeDelayErrorObservable$ParentObserver.onNext(OperationMergeDelayError.java:210)
at rx.operators.OperationMergeDelayError$2.onSubscribe(OperationMergeDelayError.java:77)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMergeDelayError$MergeDelayErrorObservable.onSubscribe(OperationMergeDelayError.java:171)
at rx.operators.OperationMergeDelayError$1.onSubscribe(OperationMergeDelayError.java:64)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:164)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:116)
at rx.operators.OperationMap$MapObservable$1.onNext(OperationMap.java:105)
at rx.operators.SafeObserver.onNext(SafeObserver.java:102)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMap$MapObservable.onSubscribe(OperationMap.java:102)
at rx.operators.OperationMap$2.onSubscribe(OperationMap.java:76)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable.onSubscribe(OperationMerge.java:106)
at rx.operators.OperationMerge$1.onSubscribe(OperationMerge.java:56)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:483)
嵌套订阅解决方案:
import rx.Observable;
import rx.Observable.OnSubscribeFunc;
import rx.Observer;
import rx.Subscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Action1;
import rx.util.functions.Func1;
public class TestMergeDelayError {
public static Observable<String> getUserIds() {
return Observable.from(new String[]{"1", "2", "3", "4", "5", "6"});
}
public static Observable<String> getInfo(final String prefix, final String integer, final String errorNumber) {
Observable<String> observable = Observable.create(new OnSubscribeFunc<String>() {
public Subscription onSubscribe(Observer<? super String> t1) {
if (integer.contains(errorNumber)) {
t1.onError(new Exception());
} else {
t1.onNext(prefix + integer);
t1.onCompleted();
}
return Subscriptions.empty();
}
});
return observable;
}
public static void main(String[] args) {
Observable<String> userIdObservable = getUserIds();
userIdObservable.subscribe(new Action1<String>() {
public void call(String t1) {
Observable<String> info1 = getInfo("1::: ", t1, "2");
Observable<String> info2 = getInfo("2::: ", t1, "3");
Observable.merge(info1, info2).subscribe(new Action1<String>() {
public void call(String t1) {
System.out.println(t1);
}
}, new Action1<Throwable>() {
public void call(Throwable t1) {
t1.printStackTrace();
}
},
new Action0() {
public void call() {
System.out.println("onComplete");
}
});
}
});
}
}
输出:
1::: 1
2::: 1
onComplete
java.lang.Exception
at TestMergeDelayError$1.onSubscribe(TestMergeDelayError.java:28)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:164)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:116)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable.onSubscribe(OperationMerge.java:106)
at rx.operators.OperationMerge$1.onSubscribe(OperationMerge.java:56)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:483)
at TestMergeDelayError$2.call(TestMergeDelayError.java:47)
at TestMergeDelayError$2.call(TestMergeDelayError.java:42)
at rx.Observable$2.onNext(Observable.java:381)
at rx.operators.SafeObserver.onNext(SafeObserver.java:102)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:367)
at TestMergeDelayError.main(TestMergeDelayError.java:42)
1::: 3
java.lang.Exception
at TestMergeDelayError$1.onSubscribe(TestMergeDelayError.java:28)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:164)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:116)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable.onSubscribe(OperationMerge.java:106)
at rx.operators.OperationMerge$1.onSubscribe(OperationMerge.java:56)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:483)
at TestMergeDelayError$2.call(TestMergeDelayError.java:47)
at TestMergeDelayError$2.call(TestMergeDelayError.java:42)
at rx.Observable$2.onNext(Observable.java:381)
at rx.operators.SafeObserver.onNext(SafeObserver.java:102)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:367)
at TestMergeDelayError.main(TestMergeDelayError.java:42)
1::: 4
2::: 4
onComplete
1::: 5
2::: 5
onComplete
1::: 6
2::: 6
onComplete
正如您所看到的，只有失败的用户 ID 停止了它们的单独处理，但其他用户 ID 得到了处理.
As you can see only the individual userids that failed stopped their individual processing but the rest of the userids were processed.
只是寻求建议，看看这个解决方案是否有意义，如果没有，最佳实践是什么.
Just looking for advice, see if this solution makes sense and if not what the best practice is.
谢谢，亚历克斯
推荐答案
最佳实践是使用 mergeDelayError( ) 将多个 Observable 合并为一个，允许无错误的 Observable 在传播错误之前继续.
The best practice is to use mergeDelayError( ) that combine multiple Observables into one, allowing error-free Observables to continue before propagating errors.
mergeDelayError 的行为很像 merge.例外是当被合并的 Observable 之一以 onError 通知终止时.如果合并发生这种情况，合并的 Observable 将立即发出 onError 通知并终止.另一方面，mergeDelayError 将推迟报告错误，直到它给任何其他非错误产生的 Observables 提供它正在合并的机会以完成发出它们的项目，并且它将自己发出这些，并且只会终止当所有其他合并的 Observable 完成时的 onError 通知.
mergeDelayError behaves much like merge. The exception is when one of the Observables being merged terminates with an onError notification. If this happens with merge, the merged Observable will immediately issue an onError notification and terminate. mergeDelayError, on the other hand, will hold off on reporting the error until it has given any other non-error-producing Observables that it is merging a chance to finish emitting their items, and it will emit those itself, and will only terminate with an onError notification when all of the other merged Observables have finished.
这篇关于处理 onError 和继续处理的最佳实践的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
