gpt4 book ai didi

java - 在 RxJava 中,如何在 Observable 之外反射(reflect)/提取故障?

转载 作者:行者123 更新时间:2023-12-01 11:45:08 26 4
gpt4 key购买 nike

我们有一个 StoreService 调用 update(key, content) 方法,该方法使用 couchbase 客户端执行 get-->change_content-->replace 操作。

作为该过程的一部分,我们使用 Observable retryWhen 来处理异常。如果重试超过最大重试次数,它只会传递异常,然后触发观察者的 onError 方法。

如果无法处理错误,我们想要做的就是从 update(key, content) 方法向调用它的 StoreService 抛出异常,但我们没有这样做。

目前我们尝试了以下方法,但没有成功:

  1. 从 onError 方法抛出异常,但不会从 Observable 中抛出异常。
  2. 抛出 RuntimeException 但效果不佳。
  3. 使用其中包含 boolean 值 isFailed 成员的 DTO:我们在 Observable 外部创建 DTO,如果出现错误,我们会收到订阅者的 onError,其中我们将 DTO 的 isFailed 设置为 true。 Observable 完成后,我们检查 DTO 是否失败,如果是,我们抛出异常。这也不起作用 - onError 中发生的更改没有传播到 Observable 之外(为什么?)

这是伪代码:

 public void update(String key, ConversationDto updateConversationDto) {

ObservableExecution observableExecution = new ObservableExecution();

Observable
.defer(... get the document from couchbase ...)
.map(... handle JSON conversion and update the document ...)
.flatMap(documentUpdate -> {
return couchbaseClient.getAsyncBucket().replace(documentUpdate);
})
.retryWhen(new RetryWithDelay(3, 200))
.subscribe(
n -> logger.debug("on next update document -> " + n.content()),
e -> {
//logger.error("failed to insert a document",e);
observableExecution.setFailure(e);
},
() -> logger.debug("on complete update document")

);
// this is never true
if (observableExecution.isFailed()) {
final Throwable e = observableExecution.getFailure();
throw new DalRuntimeException(e.getMessage(), e);
}
}

这是重试代码:

public Observable<?> call(Observable<? extends Throwable> attempts) {
return attempts
.flatMap(new Func1<Throwable, Observable<?>>() {
@Override
public Observable<?> call(Throwable errorNotification) {
if (++retryCount < maxRetries) {
// When this Observable calls onNext, the original
// Observable will be retried (i.e. re-subscribed).
logger.debug(errorNotification + " retry no. " + retryCount);
return Observable.timer(retryDelayMillis,
TimeUnit.MILLISECONDS);
}

// Max retries hit. Just pass the error along.
logger.debug(errorNotification + " exceeded max retries " + maxRetries);
return Observable.error(errorNotification);
}
});
}

非常感谢您的帮助!

最佳答案

订阅异步运行,因此 isFailed() 检查始终会在 e -> setFailure(e) 代码运行之前立即运行。

执行此操作的正确方法是从 update() 方法返回 Observable 并在 StoreService 中订阅它。这样,当您有兴趣处理成功和失败时,您就会收到通知。

关于java - 在 RxJava 中,如何在 Observable 之外反射(reflect)/提取故障?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29212740/

26 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com