处理 Spring 反应式 webclient 中的连接错误

如何解决处理 Spring 反应式 webclient 中的连接错误

我有一个 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()))

希望对您有所帮助。