使用Resilience4j实施反应式断路器 - Wenqi

22-02-09 banq

本文将重点介绍使用 Spring Cloud 断路器库 Resilience4j 实现反应式断路器。
 

为什么选择 Resilience4j?
我们可以使用两个主要库来实现断路器。Netflix Hystrix,它采用面向对象的设计,其中对外部系统的调用必须包含在HystrixCommand提供的多种功能中。但是,在 SpringOne 2019 中,Spring 宣布 Hystrix Dashboard 将从 Spring Cloud 3.1 版本中删除,这使其正式弃用。使用已弃用的库不是一个好主意。所以选择很明确,就是 Resilience4j!
Resilience4j 是一个受 Hystrix 启发的独立库,但建立在函数式编程的原则之上。Resilience4J 提供了高阶函数(装饰器),以通过断路器、速率限制器或隔板来增强任何功能接口、lambda 表达式或方法引用。
Resilience4J 的其他优点包括更精细的配置选项(例如关闭断路器模式所需的成功执行次数)和更轻的依赖项占用空间。
我们将使用两个 Spring Boot 微服务来演示如何实现响应式断路器:

  • 客户服务,充当 REST API 提供者,提供客户 CRUD 端点。
  • customer-service-client,它WebClient通过 Spring Boot Starter Webflux 库来调用 REST API。


 

步骤 1. 添加 POM 依赖项
由于我们选择WebClient使用 REST API,因此我们需要将 Spring Cloud Circuit Breaker Reactor Resilience4J 依赖项添加到我们的 REST 客户端应用程序中。

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
</dependency>
 


第2步。添加断路器配置bean
CircuitBreakerConfig类带有一组默认的断路器配置值,如果我们选择对所有的断路器使用默认的配置值,我们可以创建一个Customize bean,它被传递给ReactiveResilience4JCircuitBreakerFactory。该工厂的configureDefault方法可用于提供默认配置。示例片段如下。

@Bean
public Customizer<ReactiveResilience4JCircuitBreakerFactory> defaultCustomizer() {
    return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
            .circuitBreakerConfig(CircuitBreakerConfig.ofDefaults())
            .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(4)).build()).build());
}



如果我们选择使用自定义的配置值,我们将需要如下定义我们的bean("customer-service "只是一个REST客户端实例的样本,你可以使用你给你的REST客户端应用程序的任何实例名称)。

@Bean
public Customizer<ReactiveResilience4JCircuitBreakerFactory> customerServiceCusomtizer() {
  return factory -> {
    factory.configure(builder -> builder
      .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(2)).build())
      .circuitBreakerConfig(CircuitBreakerConfig.ofDefaults()), "customer-service");
  };
}


 
第3步. 为断路器属性添加配置
如果我们定义了我们的自定义配置豆,我们还需要在application.yml中添加断路器的配置,例如(仅是示例值,数字应根据应用的使用场景进行调整)。

resilience4j.circuitbreaker:
  instances:
    customer-service:
      failureRateThreshold: 50
      minimumNumberOfCalls: 10
      slidingWindowType: TIME_BASED
      slidingWindowSize: 10
      waitDurationInOpenState: 50s
      permittedNumberOfCallsInHalfOpenState: 3

  • failureRateThreshold。当故障率等于或大于阈值时,断路器过渡到开放状态并开始短路调用。在我们的例子中,这个值是50%,这意味着如果2个请求中有1个失败,就会达到阈值,这将使断路器进入开放状态。
  • MinimumNumberOfCalls: 这个属性确保一旦执行了最低数量的调用,就能计算出故障率。在我们的例子中,在开始计算故障率之前必须执行10个请求。
  • slidingWindowType(滑动窗口类型)。配置滑动窗口的类型,用于记录断路器关闭时的呼叫结果。滑动窗口可以是基于计数的,也可以是基于时间的。
  • slidingWindowSize(滑动窗口尺寸)。配置滑动窗口的大小,用于记录断路器关闭时的呼叫结果。
  • waitDurationInOpenState: 断路器在从开放状态过渡到半开放状态之前应该等待的时间。在我们的例子中,它是50秒。
  • permittedNumberOfCallsInHalfOpenState: 配置断路器处于半开状态时允许的呼叫数量。在我们的例子中,限制是3,这意味着在10秒的窗口中只处理3个请求。


 
第4步。实施Circuit Breaker
现在所有的配置都到位了,我们可以开始使用Circuit Breaker从客户端装饰我们的REST API调用。在下面的例子中,我们通过构造函数注入WebClient和ReactiveCircuitBreakerFactory到CustomerCientController。然后,我们使用webClient来触发对传递进来的CustomerVO和/或customerId的CRUD调用。注意 "转换 "部分,我们在ReactiveCircuitBreakerFactory的帮助下为 "customer-service "创建ReactiveCircuitBreaker实例(Rcb为ReactiveCircuitBreaker类型)。执行断路器的行是rcb.run(...)。在下面的示例控制器中,当异常被抛出时,我们为POST/GET/PUT调用返回一个空白的CustomerVO对象作为后退响应。对于DELETE调用,我们将返回传入的customerId作为回退。因此,在REST API供应商停机的情况下,我们不会得到500内部服务器错误,而是通过正确的实施Circuit Breaker,收到后备响应。

@RestController
@Slf4j
@RequiredArgsConstructor
public class CustomerClientController {

    private final WebClient webClient;
    private final ReactiveCircuitBreakerFactory reactiveCircuitBreakerFactory;

    @PostMapping("/customers")
    public Mono<CustomerVO> createCustomer(CustomerVO customerVO){
        return webClient.post()
                .uri("/customers")
                //.header("Authorization", "Bearer MY_SECRET_TOKEN")
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON)
                .body(Mono.just(customerVO), CustomerVO.class)
                .retrieve()
                .bodyToMono(CustomerVO.class)
                .timeout(Duration.ofMillis(10_000))
                .transform(it -> {
                    ReactiveCircuitBreaker rcb = reactiveCircuitBreakerFactory.create("customer-service");
                    return rcb.run(it, throwable -> Mono.just(CustomerVO.builder().build()));
                });
    }

    @GetMapping("/customers/{customerId}")
    public Mono<CustomerVO> getCustomer(@PathVariable String customerId) {
        return webClient
                .get().uri("/customers/" + customerId)
                .retrieve()
                .bodyToMono(CustomerVO.class)
                .transform(it -> {
                    ReactiveCircuitBreaker rcb = reactiveCircuitBreakerFactory.create("customer-service");
                    return rcb.run(it, throwable -> Mono.just(CustomerVO.builder().build()));
                });
    }

    @PutMapping("/customers/{customerId}")
    public Mono<CustomerVO> updateCustomer(@PathVariable String customerId, CustomerVO customerVO){
        return webClient.put()
                .uri("/customers/" + customerVO.getCustomerId())
                //.header("Authorization", "Bearer MY_SECRET_TOKEN")
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON)
                .body(Mono.just(customerVO), CustomerVO.class)
                .retrieve()
                .bodyToMono(CustomerVO.class)
                .transform(it -> {
                    ReactiveCircuitBreaker rcb = reactiveCircuitBreakerFactory.create("customer-service");
                    return rcb.run(it, throwable -> Mono.just(CustomerVO.builder().build()));
                });
    }

    @DeleteMapping("/customers/{customerId}")
    public Mono<String> deleteCustomer(@PathVariable String customerId){
        return webClient.delete()
                .uri("/customers/" + customerId)
                .retrieve()
                .bodyToMono(String.class)
                .transform(it -> {
                    ReactiveCircuitBreaker rcb = reactiveCircuitBreakerFactory.create("customer-service");
                    return rcb.run(it, throwable -> Mono.just(customerId));
                });
    }
}


 
 问题1:Swagger用户界面在Webflux中不工作
由于我们引入了Webflux库来使用WebClient,你可能会注意到你的swagger UI最初并不工作。为了让它工作,请确保以下步骤得到执行。
在pom中添加以下依赖项。

<dependency>
  <groupId>io.springfox</groupId>
  <artifactId>springfox-boot-starter</artifactId>
  <version>${springfox.version}</version>
</dependency>

如果你已经实现了注释@EnableSwagger2WebFlux,请删除该注释。
现在访问swagger的URL应该是:http://<YOUR_APP_SERVER>:<YOUR_APP_PORT>/swagger-ui/,一定要加上结尾"/"。例如,http://localhost:8100/swagger-ui/。

 

如何为没有返回类型的端点实现断路?
对于在其响应体中没有返回内容的端点,如REST API提供商中的以下端点,在REST客户端,如果我们将调用该端点的相应方法标记为返回Mono<Void>,ReactiveCircuitBreaker将无法工作。在REST API提供者关闭的情况下,你会看到500服务器错误,这完全违背了拥有Circuit Breaker的目的。

@DeleteMapping(value = "/{customerId}")
public ResponseEntity deleteCustomer(@PathVariable String customerId) throws Exception {
    customerService.deleteCustomer(customerId);
    return ResponseEntity.noContent().build();
}


在非反应式断路器的实现中,对于没有返回类型的方法,我们可以使用 "CheckedRunnable",做如下工作(示例)。

CircuitBreaker circuitBreaker = circuitBreakerFactory.create("customer-service");
CheckedRunnable runnable = () -> customerClient.deleteCustomer(customerId);
Try.run(circuitBreaker.decorateCheckedRunnable(runnable)).get();


但是,在反应式断路器中,ReactiveCircuitBreaker没有这样的接口来装饰CheckedRunnable,那么我们该怎么做?经过一些调查和实验,我注意到我们可以操纵这种端点的返回类型,使其返回一个一般的类型,如String。简单地说,如果一个端点,如DELETE调用在服务器端返回Void,我们仍然可以在客户端操纵该DELETE调用的返回类型,以返回一个简单的类型,如String,只是传回输入到该端点的String。例如,在客户端我们可以这样实现DELETE调用。

public Mono<String> deleteCustomer(@PathVariable String customerId){
    return webClient.delete()
            .uri("/customers/" + customerId)
            .retrieve()
            .bodyToMono(String.class)
            .transform(it -> {
                ReactiveCircuitBreaker rcb = reactiveCircuitBreakerFactory.create("customer-service");
                return rcb.run(it, throwable -> Mono.just(customerId));
            });
}



注意我们正在返回Mono<String>而不是Mono<Void>,而且我们在.bodyToMono(String.class)行指定了返回类型为String,这就是为什么我们可以简单地调用ReactiveCircuitBreaker的run方法来调用装饰过的反应式断路器函数。这是我能想到的唯一方法,可以解决ReactiveCircuitBreaker没有decorateCheckedRunnable方法来处理没有返回类型的方法。
完整源码:GitHub repository
 

1
猜你喜欢