Spring Cloud升级之路 - Hoxton - 4. 使用Resilience4j实现实例级别的隔离与熔断

深藏阁楼爱情的钟 2023-02-15 10:29 9阅读 0赞

## 如何不启用Hystrix ##

由于我们的入口注解类从`@SpringCloudApplication`替换成了`SpringBootApplication`，这样不会启用`Spring-Cloud-CircuitBreaker`。引入的`Hystrix`依赖也就没有效果。请参考本系列第二节： [Spring Cloud升级之路 - Hoxton - 2.入口类注解修改与OpenFeign的改造][Spring Cloud_ - Hoxton - 2._OpenFeign]

## 使用Resilience4j实现实例级别的隔离与熔断 ##

为什么需要实例级别的熔断呢？因为某个微服务可能某些实例暂时不可用，我们希望在重试的时候，暂时不再重试这些实例。默认的`Spring-Cloud-CircuitBreaker`一般实现了微服务级别的熔断，某个微服务某些实例暂时不可用但是某些实例可用的时候，就很有可能会发生整个微服务的熔断。一般在滚动发布的时候，如果操作不当，微服务级别的熔断导致这个微服务不可用，但是实际上某些实例是可用的。所以，我们需要实例级别的熔断，而不是微服务级别的。

为什么需要实例级别的线程隔离呢？防止某个实例发生问题，响应慢，阻塞了整个业务线程。

`Spring-Cloud-CircuitBreaker`里面的实现对于`resilience4j`的功能使用有限，我们想利用其更多的功能（例如线程隔离等等）。而且，`Spring-Cloud-CircuitBreaker`可以直接用来实现微服务级别的熔断，但是很难实现实例级别的熔断。主要原因是他的配置是根据微服务名称配置的，并且没有扩展，导致我们想实现的话，修改代码的地方太多了。所以我们舍弃了`Spring-Cloud-CircuitBreaker`。

比较幸运的是，`resilience4j`官方有实现自己的`spring-cloud-starter`，里面实现了他的所有功能的核心`bean`配置，很好用。我们采用这个`starter`以及相关的配置方式来实现我们的实例级别的隔离与熔断。

在引入

<dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-spring-cloud2</artifactId>
        <version>${resilience4j-spring-cloud2.version}</version>
    </dependency>

之后，会自动加载`BulkheadRegistry`，`ThreadPoolBulkheadRegistry`，`CircuitBreakerRegistry`，`RateLimiterRegistry`，`RetryRegistry`这几个`Bean`，配置这些`Bean`的分别是通过：

*  `io.github.resilience4j.bulkhead.autoconfigure.BulkheadProperties`: 前缀`resilience4j.bulkhead`
 *  `io.github.resilience4j.bulkhead.autoconfigure.ThreadPoolBulkheadProperties`:前缀`resilience4j.thread-pool-bulkhead`
 *  `io.github.resilience4j.circuitbreaker.autoconfigure.CircuitBreakerProperties`: 前缀`resilience4j.circuitbreaker`
 *  `io.github.resilience4j.ratelimiter.autoconfigure.RateLimiterProperties`:前缀`resilience4j.ratelimiter`
 *  `io.github.resilience4j.retry.autoconfigure.RetryProperties`: 前缀`resilience4j.retry`

这里主要用到的元素是：`CircuitBreaker`和`ThreadPoolBulkhead`。`CircuitBreaker`用来实现实例级别的熔断，`ThreadPoolBulkhead`用来实现实例级别的线程隔离。

## 如何配置以及如何使用 ##

`CircuitBreaker`相关的配置：[CircuitBreaker][]

`CircuitBreaker`有五种状态：CLOSED，OPEN 还有HALF\_OPEN。剩下的两种状态是人为操作，我们这里不会用到：DISABLED还有  
FORCED\_OPEN. CLOSED 代表断路器关闭，请求照常处理。OPEN 代表断路器打开，如果有请求会抛出异常:`CallNotPermittedException`。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jNL6sKBU-1591348711197)(https://zhxhash-blog.oss-cn-beijing.aliyuncs.com/spring-cloud%E5%8D%87%E7%BA%A7%E4%B9%8B%E8%B7%AF/hoxton/39cdd54-state\_machine.jpg)\]

`CircuitBreaker`使用滑动窗口统计成功失败的请求，并打开或者关闭断路器。这个滑动窗口有两种：

*  基于计数的滑动窗口：使用一个大小为 N 的环形数组，记录最近 N 个请求结果。
 *  基于计时的滑动窗口：记录最近 N 秒的请求结果

<table> 
 <thead> 
  <tr> 
   <th>配置项</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>failureRateThreshold</td> 
   <td>50</td> 
   <td>失败请求百分比，超过这个比例，<code>CircuitBreaker</code>就会变成<code>OPEN</code>状态</td> 
  </tr> 
  <tr> 
   <td>slowCallDurationThreshold</td> 
   <td>60000[ms]</td> 
   <td>慢调用时间，当一个调用慢于这个时间时，会被记录为慢调用</td> 
  </tr> 
  <tr> 
   <td>slowCallRateThreshold</td> 
   <td>100</td> 
   <td>当慢调用达到这个百分比的时候，<code>CircuitBreaker</code>就会变成<code>OPEN</code>状态</td> 
  </tr> 
  <tr> 
   <td>permittedNumberOfCallsInHalfOpenState</td> 
   <td>10</td> 
   <td>当<code>CircuitBreaker</code>处于<code>HALF_OPEN</code>状态的时候，允许通过的请求数量</td> 
  </tr> 
  <tr> 
   <td>slidingWindowType</td> 
   <td>COUNT_BASED</td> 
   <td>滑动窗口类型，<code>COUNT_BASED</code>代表是基于计数的滑动窗口，<code>TIME_BASED</code>代表是基于计时的滑动窗口</td> 
  </tr> 
  <tr> 
   <td>slidingWindowSize</td> 
   <td>100</td> 
   <td>滑动窗口大小，如果配置<code>COUNT_BASED</code>默认值100就代表是最近100个请求，如果配置<code>TIME_BASED</code>默认值100就代表是最近100s的请求。</td> 
  </tr> 
  <tr> 
   <td>minimumNumberOfCalls</td> 
   <td>100</td> 
   <td>最小请求个数。只有在滑动窗口内，请求个数达到这个个数，才会触发<code>CircuitBreaker</code>对于是否打开断路器的判断。</td> 
  </tr> 
  <tr> 
   <td>waitDurationInOpenState</td> 
   <td>60000[ms]</td> 
   <td>从<code>OPEN</code>状态变成<code>HALF_OPEN</code>状态需要的等待时间</td> 
  </tr> 
  <tr> 
   <td>automaticTransitionFromOpenToHalfOpenEnabled</td> 
   <td>false</td> 
   <td>如果设置为<code>true</code>代表是否自动从<code>OPEN</code>状态变成<code>HALF_OPEN</code>，即使没有请求过来。</td> 
  </tr> 
  <tr> 
   <td>recordExceptions</td> 
   <td>empty</td> 
   <td>异常名单，指定一个 Exception 的 list，所有这个集合中的异常或者这些异常的子类，在调用的时候被抛出，都会被记录为失败。其他异常不会被认为是失败，或者在 ignoreExceptions 中配置的异常也不会被认为是失败。默认是所有异常都认为是失败。</td> 
  </tr> 
  <tr> 
   <td>ignoreExceptions</td> 
   <td>empty</td> 
   <td>异常白名单，在这个名单中的所有异常及其子类，都不会认为是请求失败，就算在 recordExceptions 中配置了这些异常也没用。默认白名单为空。</td> 
  </tr> 
 </tbody> 
</table>

我们这里实现的默认配置是：

resilience4j.circuitbreaker:
      configs:
        default:
          # 是否向 Actuator 的 HealthIndicator 注册
          registerHealthIndicator: true
          slidingWindowSize: 10
          minimumNumberOfCalls: 5
          slidingWindowType: TIME_BASED
          permittedNumberOfCallsInHalfOpenState: 3
          automaticTransitionFromOpenToHalfOpenEnabled: true
          waitDurationInOpenState: 2s
          failureRateThreshold: 30
          recordExceptions:
            - java.lang.Exception

以上配置代表，默认情况下，所有`Exception`以及其子类都认为是失败。滑动窗口采用基于计时的，并且记录最近10秒的请求。触发断路器判断必须在10秒内至少有5个请求，在失败比例达到30%以上之后，断路器变为`OPEN`。断路器`OPEN`之后，在2秒后自动转化为`HALF_OPEN`。

`ThreadPoolBulkhead`相关的配置：[Create and configure a ThreadPoolBulkhead][]

<table> 
 <thead> 
  <tr> 
   <th>配置项</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>maxThreadPoolSize</td> 
   <td>Runtime.getRuntime().availableProcessors()</td> 
   <td>最大线程池大小</td> 
  </tr> 
  <tr> 
   <td>coreThreadPoolSize</td> 
   <td>Runtime.getRuntime().availableProcessors() - 1</td> 
   <td>核心线程池大小</td> 
  </tr> 
  <tr> 
   <td>queueCapacity</td> 
   <td>100</td> 
   <td>队列大小</td> 
  </tr> 
  <tr> 
   <td>keepAliveDuration</td> 
   <td>20[ms]</td> 
   <td>线程存活时间</td> 
  </tr> 
 </tbody> 
</table>

我们这里实现的默认配置是：

resilience4j.thread-pool-bulkhead:
      configs:
        default:
          maxThreadPoolSize: 50
          coreThreadPoolSize: 10
          queueCapacity: 1

## 与Open-Feign粘合 ##

我们需要在`FeignClient`被调用，选取好要发送请求的实例之后，加入`CircuitBreaker`和`ThreadPoolBulkhead`。也就是，我们需要拿到本次请求调用的实例，以及微服务名称，加载对应的`CircuitBreaker`和`ThreadPoolBulkhead`，包装调用请求，之后执行调用。

`FeignClient`的核心实现，根据`org.springframework.cloud.openfeign.loadbalancer.DefaultFeignLoadBalancerConfiguration`可以知道是`org.springframework.cloud.openfeign.loadbalancer.FeignBlockingLoadBalancerClient`：

@Bean
    @ConditionalOnMissingBean
    public Client feignClient(BlockingLoadBalancerClient loadBalancerClient) {
    	return new FeignBlockingLoadBalancerClient(new Client.Default(null, null),
    			loadBalancerClient);
    }

查看`FeignBlockingLoadBalancerClient`的源码：

@Override
    public Response execute(Request request, Request.Options options) throws IOException {
    	final URI originalUri = URI.create(request.url());
    	//微服务名称
    	String serviceId = originalUri.getHost();
    	Assert.state(serviceId != null,
    			"Request URI does not contain a valid hostname: " + originalUri);
    	//从负载均衡器选择一个实例
    	ServiceInstance instance = loadBalancerClient.choose(serviceId);
    	if (instance == null) {
    		String message = "Load balancer does not contain an instance for the service "
    				+ serviceId;
    		if (LOG.isWarnEnabled()) {
    			LOG.warn(message);
    		}
    		return Response.builder().request(request)
    				.status(HttpStatus.SERVICE_UNAVAILABLE.value())
    				.body(message, StandardCharsets.UTF_8).build();
    	}
    	//修改原url
    	String reconstructedUrl = loadBalancerClient.reconstructURI(instance, originalUri)
    			.toString();
    	//构建新的Request
    	Request newRequest = Request.create(request.httpMethod(), reconstructedUrl,
    			request.headers(), request.body(), request.charset(),
    			//这个RequestTemplate可以拿到微服务名称
    			request.requestTemplate());
    	return delegate.execute(newRequest, options);
    }

所以，我们可以通过继承`FeignBlockingLoadBalancerClient`替换默认的实现，来代理调用请求。但是因为`sleuth`的存在以及其中的小`bug`导致`RequestTemplate`丢失，让我们拿不到微服务名称，这个可以参考我提的 PR：[replace method for deprecation and keep reference of requestTemplate][].但是在`Hoxton`版本不会合并了，所以需要我们建立同名同路径类进行替换：[org.springframework.cloud.sleuth.instrument.web.client.feign.TracingFeignClient][]

Request build() {
        if (headers == null) {
            return delegate;
        }
        String url = delegate.url();
        byte[] body = delegate.body();
        Charset charset = delegate.charset();
        //保留requestTemplate
        return Request.create(delegate.httpMethod(), url, headers, body, charset, delegate.requestTemplate());
    }

之后，我们实现带`CircuitBreaker`和`ThreadPoolBulkhead`的`FeignBlockingLoadBalancerClient`，并优化其中的`HttpClient`:

@Bean
    public HttpClient getHttpClient() {
        // 长连接保持30秒
        PoolingHttpClientConnectionManager pollingConnectionManager = new PoolingHttpClientConnectionManager(5, TimeUnit.MINUTES);
        // 总连接数
        pollingConnectionManager.setMaxTotal(1000);
        // 同路由的并发数
        pollingConnectionManager.setDefaultMaxPerRoute(1000);
    
        HttpClientBuilder httpClientBuilder = HttpClients.custom();
        httpClientBuilder.setConnectionManager(pollingConnectionManager);
        // 保持长连接配置，需要在头添加Keep-Alive
        httpClientBuilder.setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy());
        return httpClientBuilder.build();
    }
    
    @Bean
    public FeignBlockingLoadBalancerClient feignBlockingLoadBalancerCircuitBreakableClient(HttpClient httpClient, BlockingLoadBalancerClient loadBalancerClient, BulkheadRegistry bulkheadRegistry, ThreadPoolBulkheadRegistry threadPoolBulkheadRegistry, CircuitBreakerRegistry circuitBreakerRegistry, RateLimiterRegistry rateLimiterRegistry, RetryRegistry retryRegistry, Tracer tracer) {
        return new FeignBlockingLoadBalancerClient(new CircuitBreakableClient(
                httpClient,
                bulkheadRegistry,
                threadPoolBulkheadRegistry,
                circuitBreakerRegistry,
                rateLimiterRegistry,
                retryRegistry,
                tracer),
                loadBalancerClient);
    }
    
    @Log4j2
    public static class CircuitBreakableClient extends feign.httpclient.ApacheHttpClient {
        private final BulkheadRegistry bulkheadRegistry;
        private final ThreadPoolBulkheadRegistry threadPoolBulkheadRegistry;
        private final CircuitBreakerRegistry circuitBreakerRegistry;
        private final RateLimiterRegistry rateLimiterRegistry;
        private final RetryRegistry retryRegistry;
        private final Tracer tracer;
    
        public CircuitBreakableClient(HttpClient httpClient, BulkheadRegistry bulkheadRegistry, ThreadPoolBulkheadRegistry threadPoolBulkheadRegistry, CircuitBreakerRegistry circuitBreakerRegistry, RateLimiterRegistry rateLimiterRegistry, RetryRegistry retryRegistry, Tracer tracer) {
            super(httpClient);
            this.bulkheadRegistry = bulkheadRegistry;
            this.threadPoolBulkheadRegistry = threadPoolBulkheadRegistry;
            this.circuitBreakerRegistry = circuitBreakerRegistry;
            this.rateLimiterRegistry = rateLimiterRegistry;
            this.retryRegistry = retryRegistry;
            this.tracer = tracer;
        }
    
        @Override
        public Response execute(Request request, Request.Options options) throws IOException {
            String serviceName = request.requestTemplate().feignTarget().name();
            URL url = new URL(request.url());
            String instanceId = serviceName + ":" + url.getHost() + ":" + url.getPort();
    
            //每个实例一个resilience4j熔断记录器，在实例维度做熔断，所有这个服务的实例共享这个服务的resilience4j熔断配置
            ThreadPoolBulkhead threadPoolBulkhead;
            CircuitBreaker circuitBreaker;
            try {
                threadPoolBulkhead = threadPoolBulkheadRegistry.bulkhead(instanceId, serviceName);
            } catch (ConfigurationNotFoundException e) {
                threadPoolBulkhead = threadPoolBulkheadRegistry.bulkhead(instanceId);
            }
            try {
                circuitBreaker = circuitBreakerRegistry.circuitBreaker(instanceId, serviceName);
            } catch (ConfigurationNotFoundException e) {
                circuitBreaker = circuitBreakerRegistry.circuitBreaker(instanceId);
            }
            //保持traceId
            Span span = tracer.currentSpan();
            Supplier<CompletionStage<Response>> completionStageSupplier = ThreadPoolBulkhead.decorateSupplier(threadPoolBulkhead,
                    CircuitBreaker.decorateSupplier(circuitBreaker, () -> {
                        try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) {
                            log.info("call url: {} -> {}", request.httpMethod(), request.url());
                            Response execute = super.execute(request, options);
                            if (execute.status() != HttpStatus.OK.value()) {
                                execute.close();//需要关闭，否则返回码不为200抛异常连接不会回收导致连接池耗尽
                                throw new ResponseWrapperException(execute.toString(), execute);
                            }
                            return execute;
                        } catch (Exception e) {
                            throw new ResponseWrapperException(e.getMessage(), e);
                        }
                    })
            );
    
            try {
                return Try.ofSupplier(completionStageSupplier).get().toCompletableFuture().join();
            } catch (CompletionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof ResponseWrapperException) {
                    ResponseWrapperException responseWrapperException = (ResponseWrapperException) cause;
                    if (responseWrapperException.getResponse() != null) {
                        return (Response) responseWrapperException.getResponse();
                    }
                }
                throw new ResponseWrapperException(cause.getMessage(), cause);
            }
        }
    }

这样，我们就粘合了Open-Feign，加入了需要的基于实例的熔断和线程隔离

## 在 Spring Cloud Gateway 中实现基于实例的熔断 ##

Spring Cloud Gateway 不用做线程隔离，因为 reactor 框架不是同步框架，某个实例发生阻塞对它影响不至于很大。我们只在其中做实例级别的熔断，每个微服务都会做这个实例级别的熔断。所以需要加一个`GlobalFilter`。这个熔断需要在负载均衡器选择实例并重写调用 url 之后，也就是在 `ReactiveLoadBalancerClientFilter` 之后。

由于Spring Cloud Gateway 基于 reactor 做了异步，所以我们需要 `spring-cloud-starter-circuitbreaker-reactor-resilience4j`里面的`CircuitBreakerOperator`将断路器转换成 reactor 兼容的`Publisher`。所以加入了依赖：

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

我们只为了用`CircuitBreakerOperator`做个转换而已，其他的并没有用到。

[InstanceCircuitBreakerFilter][]

/**
     * 实例级别的断路器
     */
    @Log4j2
    @Component
    public class InstanceCircuitBreakerFilter implements GlobalFilter, Ordered {
    
        private final CircuitBreakerRegistry circuitBreakerRegistry;
        ObjectMapper objectMapper = new ObjectMapper();
    
    
        public InstanceCircuitBreakerFilter(CircuitBreakerRegistry circuitBreakerRegistry) {
            this.circuitBreakerRegistry = circuitBreakerRegistry;
        }
    
        @Override
        public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
            URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
            ServerHttpRequest request = exchange.getRequest();
            String serviceName = request.getHeaders().getFirst(CommonConstant.SERVICE_NAME);
            String instanceId = url.getHost() + url.getPort();
            CircuitBreaker circuitBreaker;
            try {
                //使用实例id新建或者获取现有的CircuitBreaker,使用serviceName获取配置
                circuitBreaker = circuitBreakerRegistry.circuitBreaker(instanceId, serviceName);
            } catch (ConfigurationNotFoundException e) {
                circuitBreaker = circuitBreakerRegistry.circuitBreaker(instanceId);
            }
    
            try {
                log.info("try to send request to: {}: stats: {}", url, objectMapper.writeValueAsString(circuitBreaker.getMetrics()));
            } catch (JsonProcessingException e) {
            }
    
            return chain.filter(exchange).transform(CircuitBreakerOperator.of(circuitBreaker));
        }
    
        @Override
        public int getOrder() {
            try {
                //必须在负载均衡器选择实例并重写调用 url 之后，也就是在 `ReactiveLoadBalancerClientFilter` 之后。
                return (Integer) ReactiveLoadBalancerClientFilter.class.getDeclaredField("LOAD_BALANCER_CLIENT_FILTER_ORDER").get(null) + 1;
            } catch (Exception e) {
                return 10151;
            }
        }
    
    }

[Spring Cloud_ - Hoxton - 2._OpenFeign]: https://blog.csdn.net/zhxdick/article/details/106542110
[CircuitBreaker]: https://resilience4j.readme.io/docs/circuitbreaker
[Create and configure a ThreadPoolBulkhead]: https://resilience4j.readme.io/docs/bulkhead
[replace method for deprecation and keep reference of requestTemplate]: https://github.com/spring-cloud/spring-cloud-sleuth/pull/1622
[org.springframework.cloud.sleuth.instrument.web.client.feign.TracingFeignClient]: https://github.com/HashZhang/spring-cloud-scaffold/blob/master/spring-cloud-hoxton/spring-cloud-hoxton-service-consumer/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/feign/TracingFeignClient.java
[InstanceCircuitBreakerFilter]: https://github.com/HashZhang/spring-cloud-scaffold/blob/master/spring-cloud-hoxton/spring-cloud-hoxton-api-gateway/src/main/java/com/github/hashjang/hoxton/api/gateway/filter/InstanceCircuitBreakerFilter.java