spring-boot - 使用 reactiveFeignClient 和 CircuitBreaker 进行错误处理

Question

我们正在使用响应式(Reactive)伪装客户端 (com.playtika.reactivefeign:feign-reactor-spring-cloud-starter:3.2.0)

断路器版本:org.springframework.cloud:spring-cloud-starter-circuitbreaker-reactor-resilience4j:2.1.0

和 spring boot 应用程序版本 org.springframework.boot' version '2.6.6

当我们从响应式(Reactive)伪装客户端收到错误(例如 404 错误)时

@ReactiveFeignClient(name = "someRestClient", url = "${react-gpi-service.url}",configuration = AuthConfigurationsomeRestClient.class, fallbackFactory = someRestClienttFallbackFactory.class)
@Profile("!test")
public interface someRestClient {
@PostMapping(value = "/v2/{entity}/any", produces = MediaType.ALL_VALUE, consumes = MediaType.APPLICATION_JSON_VALUE)
 Mono<String> any(@PathVariable(value = "entity")

它遍历错误解码器以检查是否应该重试

@Slf4j
@RequiredArgsConstructor
public class RetryableErrorDecoder implements ErrorDecoder {

    private static ErrorDecoder defaultErrorDecoder = new Default();
    private final String clientName;

    public Exception decode(String methodKey, Response response) {
        String body = "";
        try {
            body = IOUtils.toString(response.body().asInputStream(), StandardCharsets.UTF_8);
        } catch (Exception e) {
            log.error("failed to parse error response body", e);
        }

        log.error("In RetryableErrorDecoder, got an error from {}. status: {}, body: {}, reason: {}, request: {}",
                clientName, response.status(), body, response.reason(), response.request());

        if (response.status() == HttpStatusCodes.STATUS_CODE_SERVICE_UNAVAILABLE ||
                response.status() == HttpStatusCodes.STATUS_CODE_BAD_GATEWAY) {
            log.warn("Retry on error 503 or 502");
            return createRetryableException(response, "Service Unavailable 503, 502");
        } else {
            Exception decode = defaultErrorDecoder.decode(methodKey, response);
            if (decode instanceof FeignException &&
                    decode.getMessage().contains("authorizing")) {
                log.warn("Retry on {}", decode.getMessage());
                return createRetryableException(response, "Service authorizing problem");
            }
            return decode;
        }
    }

    private Exception createRetryableException(Response response, String message) {
        return new RetryableException(
                response.status(),
                message,
                response.request().httpMethod(),
                null,
                null,
                response.request());
    }
}

之后进入 Circuit beaker 谓词

public class someFailurePredicate  implements Predicate<Throwable> {

    @Override
    public boolean test(Throwable throwable) {
        return throwable instanceof ThirdPartyException
                || throwable instanceof ReadTimeoutException
                || throwable instanceof OutOfRetriesException;
    }
}

然后进入 fallBackFactory 机制，因为断路器需要 fallback 方法，所以断路器谓词再次被激活。

@Component
public class someRestClientFallbackFactory implements FallbackFactory<someRestClient> {

    @Override
    public someRestClient apply(Throwable throwable) {
        return new someRestClientFallback(throwable);
    }
}

public class someRestClientFallback implements someRestClient {

    private final Throwable cause;

    public someClientFallback(Throwable cause) {
        this.cause = cause;
    }

    public Mono<String> performSearchRequest(String entity,
                                                     ) {
        return Mono.error(cause);
    }
}

因为我们有 2 种错误处理机制，电路谓词调用两次并复制错误。

我尝试将重试机制(错误解码器)移至回退方法，但回退工厂方法接受可抛出的并且 reactiveFeignClientException 没有状态代码，因此很难确定我们是否应该进行重试。

如果我删除后备方法，我会收到此错误消息:

org.springframework.cloud.client.circuitbreaker.NoFallbackAvailableException: No fallback available.

我们需要添加它，但是我们有两个机制和一个重复的断路器谓词计数

Answer 1

Reactive Feign Client 默认启用自己的 CB，可以通过将 reactive.feign.circuit.breaker.enabled 设置为 false 来禁用它 - https://github.com/PlaytikaOSS/feign-reactive/blob/develop/feign-reactor-spring-configuration/README.md