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

Question

我有一个spring webclient对外部服务进行http调用，并得到反应式断路器工厂（resilience4J impl）的支持。当客户端建立连接并响应失败（任何内部服务器或4XX错误）时，WebClient和断路器的行为符合预期。但是，如果客户端无法建立连接，无论是连接被拒绝还是未知主机，它都会开始崩溃。

1. 我似乎无法在webClient中捕获错误消息并触发断路器。
2. 断路器永远不会打开并抛出TimeoutException。
   java.util.concurrent.TimeoutException：在“电路断路器”中1000毫秒内没有观察到任何项目或终端信号（并且没有配置回退）。

来自Web客户端的错误。
io.netty.channel.AbstractChannel$AnnotatedConnectException：连接拒绝：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, 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()));

}

Answer 1

我将对您的解决方案提出以下建议：

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()))

希望这有帮助。

Answer 2

我通过添加onError继续块并重新抛出异常作为在我的断路器代码中处理的自定义来修复它。

.onErrorContinue(
        (throwable, o) -> {
          log.info("throwable => {}", throwable.toString());
          if (throwable instanceof ReadTimeoutException || throwable instanceof ConnectException) {
            throw new CircuitBreakerOpenException();
          }
        })