问题描述
我有一个 spring 网络客户端,它对外部服务进行 http 调用,并由反应式断路器工厂 (resilience4J impl) 提供支持。当客户端建立连接并响应失败(任何内部服务器或 4XX 错误)时,WebClient 和断路器的行为符合预期。但是,如果客户端无法建立连接,无论是连接被拒绝还是未知主机,它都会开始崩溃。
- 我似乎无法在网络客户端中捕获错误消息并触发断路器。
- 断路器永远不会打开并抛出 TimeoutException。
java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 1000ms in 'circuitBreaker' (and no fallback has been configured)
。
来自 Web 客户端的错误。io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: localhost/127.0.0.1:9000
。
这是我的代码。我也粘贴了错误来源。我试图将 ConnectException 映射到我的自定义异常,以便断路器拾取,但是,它没有用。有人可以帮助我在没有远程服务器响应的情况下处理错误吗?
public Mono<String> toSink(
Envelope envelope,ConsumerConfiguration webClientConfiguration) {
return getWebClient()
.post()
.uri(
uriBuilder -> {
if (webClientConfiguration.getPort() != null) {
uriBuilder.port(webClientConfiguration.getPort());
}
return uriBuilder.path(webClientConfiguration.getServiceURL()).build();
})
.headers(
httpHeaders ->
webClientConfiguration.getHttpHeaders().forEach((k,v) -> httpHeaders.add(k,v)))
.bodyValue(envelope.toString())
.retrieve()
.bodyToMono(Map.class)
// Convert 5XX internal server error and throw CB exception
.onErrorResume(
throwable -> {
log.error("Inside the error resume callback of webclient {}",throwable.toString());
if (throwable instanceof WebClientResponseException) {
WebClientResponseException r = (WebClientResponseException) throwable;
if (r.getStatusCode().is5xxServerError()) {
return Mono.error(new CircuitBreakerOpenException());
}
}
return Mono.error(new CircuitBreakerOpenException());
})
.map(
map -> {
log.info("Response map:{}",Any.wrap(map).toString());
return Status.SUCCESS.name();
})
.transform(
it -> {
ReactiveCircuitBreaker rcb =
reactiveCircuitBreakerFactory.create(
webClientConfiguration.getCircuitBreakerId());
return rcb.run(
it,throwable -> {
/// "Did not observe any item or terminal signal within 1000ms.. " <--- Error here
log.info("throwable in CB {}",throwable.toString());
if (throwable instanceof CygnusBusinessException) {
return Mono.error(throwable);
}
return Mono.error(
new CircuitBreakerOpenException(
throwable,new CygnusContext(),null,null));
});
})
///io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: localhost/127.0.0.1:9000 <-- Error prints here
.onErrorContinue((throwable,o) -> log.error(throwable.toString()))
.doOnError(throwable -> log.error("error from webclient:{}",throwable.toString()));
}
解决方法
我通过添加一个 onErrorContinue
块并重新抛出异常作为在我的断路器代码中处理的自定义来修复它。
.onErrorContinue(
(throwable,o) -> {
log.info("throwable => {}",throwable.toString());
if (throwable instanceof ReadTimeoutException || throwable instanceof ConnectException) {
throw new CircuitBreakerOpenException();
}
})
,
我会就您的解决方案提出以下建议:
1- onErrorContinue
的另一种变体接受谓词,因此您可以定义此运算符将应用于哪些异常 - Docs
2- 返回 Mono.error 而不是从 Mono/Flux 运算符抛出 RuntimeExceptions
。这个其他的 stackoverflow 答案很好地涵盖了这一点 - Stackoverflow
3- 使用副作用运算符 (doOn*
) 执行日志记录
.doOnError(throwable -> log.info("throwable => {}",throwable.toString()))
.onErrorResume(throwable -> throwable instanceof ReadTimeoutException || throwable instanceof ConnectException,t -> Mono.error(new CircuitBreakerOpenException()))
希望对您有所帮助。