我正在尝试使用 java Rx(版本 1)重试。
我想要执行 retryWhen
而不是简单的 retry()
因为我希望在达到限制时返回具有特定值的可观察值,而不是仅仅抛出异常.
所以,检查这个 https://blog.danlew.net/2016/01/25/rxjavas-repeatwhen-and-retrywhen-explained/还有这个Catch error if retryWhen:s retries runs out我能够构建一些有助于实现我的目标的东西。
// this is only to simulate the real method that will possibly throw an exception
public static Observable<String> test() {
Observable<String> var = Observable.error(new IOException());
return var;
}
Observable<String> test = test().retryWhen(attempts -> {
return attempts.zipWith(Observable.range(1, 3), (throwable, attempt) -> {
if (attempt == 3) {
LOG.info("attempting");
return Observable.just("completed with error");
} else {
return attempt;
}
});
});
test.doOnError(x -> System.out.println("do on error message")).subscribe(s -> {
System.out.println(s);
});
当我在本地运行此命令时,我看到尝试 3 次的日志记录(如预期)。
我没有看到println“执行错误消息”
(如预期)
但是我没有看到我所期待的完成但有错误
,这让我怀疑我是否真的返回了我想要的可观察值,我做错了什么?
我也不明白为什么它允许我在 zipWith 中返回一个可观察值和一个整数。有什么想法吗?
并且,是否可以从我自己的可观察定义中抛出异常/错误?像这样的东西:
Observable<String> test = test().retry(3).map(value -> {
// some logic to define what to do
Observable.error(new Exception("error");
});
最佳答案
首先,
I also don't understand why it allows me to return an observable and an integer inside zipWith.
zipWith 中 lambda 的签名是 (Throwable, Integer) -> Object
意味着任何内容都是有效的返回,因为它是 Object 的子级
。出现这种情况是因为该函数定义了如何组合两个对象(在本例中为 Throwable 和 Integer,并且任何 Object 都是有效的组合(或缺少组合)。
回到你的主要问题。记住 retryWhen
实际上在做什么非常重要。这有点难以理解(至少对我来说),但基本上每当 retryWhen
主体中的观察者发出时,都会导致上游 Observable
被重新订阅到。这不会控制下游排放。
来自 docs 的示例(RxJava 2 片段,但这种观点仍然适用)显示:
Observable.create((ObservableEmitter<? super String> s) -> {
System.out.println("subscribing");
s.onError(new RuntimeException("always fails"));
}).retryWhen(attempts -> {
return attempts.zipWith(Observable.range(1, 3), (n, i) -> i).flatMap(i -> {
System.out.println("delay retry by " + i + " second(s)");
return Observable.timer(i, TimeUnit.SECONDS);
});
}).blockingForEach(System.out::println);
在此示例中,retryWhen
block 中的返回控制我们何时重新订阅初始源。在本例中,我们表示希望将重新订阅延迟 i
秒。
考虑到这一点,retryWhen
可能不是您最初寻求的解决方案。另一种解决方案可能是使用retry
,无论您想尝试订阅多少次(或者使用retryWhen
,如果您想要更自定义的重新订阅),然后使用onErrorResumeNext
。另请参阅this .
举个例子:
Observable.create((ObservableEmitter<String> s) -> s.onError(new RuntimeException("always fails")))
.retry(3)
.onErrorResumeNext(throwable -> {
return Observable.just("hi");
})
.subscribe(System.out::println, System.out::println);
结果输出是hi
。这里的关键是 onErrorResumeNext
允许我们将发出的异常转换为其他内容。几乎就像异常情况的 map
。
关于java - 我可以使用 retryWhen 并在达到限制时返回可观察值吗?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/56826610/