Spring Cloud Circuit Breaker

2021-12-28 12:40:40 浏览数 (1)

支持实现

  • Netfix Hystrix
代码语言:javascript复制
<!--SpringCloud依赖-->
org.springframework.cloud:spring-cloud-starter-netflix-hystrix
  • Resilience4J
代码语言:javascript复制
<!-- SpringCloud依赖(Resilience4j的方式) -->
 org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j

<!-- SpringCloud依赖(Reactive Resilience4J的方式) --->
org.springframework.cloud:spring-cloud-starter-circuitbreaker-reator-resilience4j

官方介绍:

Resilience4j是一款轻量级,易于使用的容错库,其灵感来自于Netflix Hystrix,专为Java8和函数式编程而设计。轻量级,因为库只使用了Vavr,它没有任何其他外部依赖下。相比之下,Netflix HystrixArchaius具有编译依赖性,Archaius具有更多的外部库依赖性,例如GuavaApache Commons Configuration

多种支持:

提供对Spring Boot2.XSpring Cloud以及 RXJAVA2的支持等

环境要求:

JDK1.8

配置属性

默认值

描述

failureRateThreshold

50(%)

当故障率等于或大于阈值时,CircuitBreaker转换为打开状态并开始短路呼叫。

slowCallRateThreshold

100

配置百分比阈值。当通话时长大于10分钟时,CircuitBreaker会认为通话缓慢slowCallDurationThreshold当慢速呼叫的百分比等于或大于阈值时,CircuitBreaker转换为打开并开始短路呼叫。

slowCallDurationThreshold

60(s)

配置持续时间阈值,在该阈值以上,呼叫将被视为慢速并增加慢速呼叫的速率。

permittedNumberOfCallsInHalfOpenState

10

配置当CircuitBreaker半开时允许的呼叫数。

slideWindowType

COUNT_BASED

配置滑动窗口的类型,该窗口用于在CircuitBreaker关闭时记录呼叫结果。 滑动窗口可以基于计数或基于时间。如果滑动窗口为COUNT_BASED,slidingWindowSize则记录并汇总最近的调用。 如果滑动窗口是TIME_BASED,则slidingWindowSize记录并汇总最近几秒的调用。

slideWindowSize

100

配置滑动窗口的大小,该窗口用于记录CircuitBreaker关闭时的呼叫结果。

minimumNumberOfCalls

10

配置CircuitBreaker可以计算错误率之前所需的最小呼叫数(每个滑动窗口时段)。例如,如果minimumNumberOfCalls为10,则在计算失败率之前,必须至少记录10个呼叫。如果仅记录了9个呼叫,则即使所有9个呼叫均失败,CircuitBreaker也不会转换为打开。

waitDurationInOpenState

60(s)

从打开状态转为半开状态等待的时间

recordExceptions

empty

需要记录的异常列表

ignoreExceptions

empty

需要忽略的异常列表

recordException

throwable-> true默认情况下,所有异常都记录为失败。

用于评估是否应将异常记录为失败。如果异常应计为失败,则必须返回true。如果异常应被视为成功,则必须返回false,除非该异常被显式忽略ignoreExceptions。

ignoreException

throwable->false默认情况下,不会忽略任何异常。

用于评估是否应忽略异常,并且该异常既不算作失败也不算成功。如果应忽略异常,则必须返回true。否则必须返回false。

automaticTransitionFromOpenToHalfOpenEnabled

false

如果置为true,当等待时间结束会自动由打开变为半开,若置为false,则需要一个请求进入来触发熔断器状态转换

代码语言:javascript复制
CircuitBreakerConfig cfg = CircuitBreakerConfig
                .custom()
              .failureRateThreshold(50) // 错误率,这个是根据滑动窗口大小决定的,e.g. windowSize = 2,failureRate=50% 那么,当出现一个错误的时候即为失败
                .waitDurationInOpenState(Duration.ofMillis(1000))
              .permittedNumberOfCallsInHalfOpenState(2)
                .slidingWindowSize(2) // 滑动窗口大小
              .recordExceptions(RuntimeException.class) // 当出现列表中的异常类型时记录
                .build();
  • RateLimiter
    • 默认实现 - io.github.resilience4j.ratelimiter.internal.AtomicRateLimiter
      • State -io.github.resilience4j.ratelimiter.internal.AtomicRateLimiter.State
        • activeCycle - 上一次调用使用的周期号
        • activePermissions- 上次调用后的可用权限计数。 如果保留某些权限,则可以为负
        • nanosToWait- 等待上一次呼叫的等待许可的纳秒数
代码语言:javascript复制
  <!-- 需要添加的依赖 -->
  <dependency>
      <groupId>io.github.resilience4j</groupId>
      <artifactId>resilience4j-ratelimiter</artifactId>
      <version>${resilience4j.version}</version>
  </dependency>
配置项

配置属性

默认值

描述

timeoutDuration

5 [s]

默认等待权限持续时间

limitRefreshPeriod

500 [ns]

限制刷新的时间段。在每个时间段之后,速率限制器将其权限计数重新设置为limitForPeriod值

limitForPeriod

50

限制刷新期间段可用的权限数

  • Bulkhead
代码语言:javascript复制
<!-- 需要添加的依赖 -->
<dependency>
     <groupId>io.github.resilience4j</groupId>
     <artifactId>resilience4j-bulkhead</artifactId>
     <version>${resilience4j.version}</version>
</dependency>

实现方式

  • SemaphoreBulkhead - 使用信号量

应该应用于多线程环境或者I/O密集型的场景下,基于semaphore,但与histrix不同,他不提供'shadow'线程池选项, 客户端应确保正确的线程池大小与bulkhead 配置保持一致。

配置项:

配置属性

默认值

描述

maxConcurrentCalls

25

允许的最大并行执行量

maxWaitDuration

0

尝试进入饱和的bulkhead时,应阻塞线程的最长时间。

代码语言:javascript复制
  BulkheadConfig config = BulkheadConfig.custom()
                .maxConcurrentCalls(5)
                .maxWaitDuration(Duration.ofMillis(1))
                .build();
        BulkheadRegistry registry = BulkheadRegistry.of(config);
        Bulkhead bulkhead = registry.bulkhead("Bulkhead");
        for (int i = 0; i<10;i  ){
            Supplier<String> decoratedSupplier = Bulkhead
                    .decorateSupplier(bulkhead, CircuitBreakerService::say);
            String result = Try.ofSupplier(decoratedSupplier)
                    .recover(throwable -> "bulkhead").get();
            System.out.println(result);
        }
  • FixedThreadPoolBulkhead - 使用有界队列和固定大小的线程池

配置项

配置属性

默认值

描述

maxThreadPoolSize

Runtime.getRuntime() .availableProcessors()

最大线程池大小

coreThreadPoolSize

Runtime.getRuntime() .availableProcessors() - 1

核心线程池大小

queueCapacity

100

队列的容量

keepAliveDuration

20(ms)

当线程总数大于核心线程数时,空闲线程存活的最大时间

代码语言:javascript复制
ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
                .maxThreadPoolSize(2)
                .coreThreadPoolSize(1)
                .queueCapacity(1)
                .build();
        ThreadPoolBulkheadRegistry registry = ThreadPoolBulkheadRegistry.of(config);
        ThreadPoolBulkhead bulkhead = registry.bulkhead("ThreadPoolBulkhead");
        ThreadPoolBulkhead bulkhead2 = registry.bulkhead("ThreadPoolBulkhead2");
        ThreadPoolBulkhead bulkhead3 = registry.bulkhead("ThreadPoolBulkhead3");
        Callable<CompletionStage<String>> call =
                ThreadPoolBulkhead.decorateCallable(bulkhead, () -> CircuitBreakerService.say());

        Callable<CompletionStage<String>> call2 =
                ThreadPoolBulkhead.decorateCallable(bulkhead2, () -> CircuitBreakerService.say());

        Callable<CompletionStage<String>> call3 =
                ThreadPoolBulkhead.decorateCallable(bulkhead3, () -> CircuitBreakerService.say());


        ExecutorService es = Executors.newFixedThreadPool(2);
        es.submit(call);
        es.submit(call2);
        es.submit(call3);
  • Retry
代码语言:javascript复制
<!-- 需要添加的依赖 -->
<dependency>
     <groupId>io.github.resilience4j</groupId>
     <artifactId>resilience4j-retry</artifactId>
     <version>${resilience4j.version}</version>
</dependency>

配置项

配置属性

默认值

描述

maxAttempts

3

最大重试次数

waitDuration

500 [ms]

重试尝试过程等待时间

intervalFunction

numOfAttempts -> waitDuration

发生故障后修改等待间隔的功能。默认情况下,等待时间保持不变。

retryOnResultPredicate

result -> false

是否需要重试,如果需要则必须置位true, 否则应置位为false

retryOnExceptionPredicate

throwable -> true

发生故障后是否需要重试,如果需要则必须置位true, 否则应置位为false

retryExceptions

empty

需要重试的异常类型列表

ignoreExceptions

empty

忽略重试的异常类型列表

代码语言:javascript复制
 RetryConfig config = RetryConfig.custom()
                .maxAttempts(2) // 最大重试次数
                .waitDuration(Duration.ofMillis(100)) // 等待时间
                .retryOnException(e -> e instanceof WebServiceException)
                .retryExceptions(IOException.class, TimeoutException.class,RuntimeException.class)
//                .ignoreExceptions(TimeoutException.class)
                .build();
        RetryRegistry registry = RetryRegistry.of(config);
        Retry retry = registry.retry("Retry");
 Function<Void , String> decorated
                = Retry.decorateFunction(retry, (s) -> CircuitBreakerService.say());
        decorated.apply(null);
  • Cache
代码语言:javascript复制
<!-- 需要添加的依赖 -->
<dependency>
     <groupId>io.github.resilience4j</groupId>
     <artifactId>resilience4j-cache</artifactId>
     <version>${resilience4j.version}</version>
</dependency>

注意:

不建议在生产中使用参考JCache 的实现,因为它会导致一些并发问题。使用EhcacheCaffeineRedissonHazelcastIgnite或其他实现。

Spring Cloud 配置 Resilience4J

1.1. 启动器

  • org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j -非反应式应用
  • org.springframework.cloud:spring-cloud-starter-circuitbreaker-reactor-resilience4j -反应式应用

1.2. 禁用自动装配

代码语言:javascript复制
spring.cloud.circuitbreaker.resilience4j.enabled=false

1.3. 默认配置

  • Resilience4JCircuitBreakerFactory
代码语言:javascript复制
@Bean
public Customizer<Resilience4JCircuitBreakerFactory> defaultCustomizer() {
    return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
            .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(4)).build())
            .circuitBreakerConfig(CircuitBreakerConfig.ofDefaults())
            .build());
}
  • ReactiveResilience4JCircuitBreakerFactory
代码语言:javascript复制
@Bean
public Customizer<Resilience4JCircuitBreakerFactory> defaultCustomizer() {
    return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
            .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(4)).build())
            .circuitBreakerConfig(CircuitBreakerConfig.ofDefaults())
            .build());
}

1.4. 特殊熔断配置

代码语言:javascript复制
@Bean
public Customizer<Resilience4JCircuitBreakerFactory> slowCustomizer() {
    return factory -> factory.configure(builder -> builder.circuitBreakerConfig(CircuitBreakerConfig.ofDefaults())
            .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(2)).build()), "slow");
}

@Bean
public Customizer<Resilience4JCircuitBreakerFactory> slowCustomizer() {
    return factory -> factory.addCircuitBreakerCustomizer(circuitBreaker -> circuitBreaker.getEventPublisher()
    .onError(normalFluxErrorConsumer).onSuccess(normalFluxSuccessConsumer), "normalflux");
}

// Reactive方式
@Bean
public Customizer<ReactiveResilience4JCircuitBreakerFactory> slowCusomtizer() {
    return factory -> {
        factory.configure(builder -> builder
        .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(2)).build())
        .circuitBreakerConfig(CircuitBreakerConfig.ofDefaults()), "slow", "slowflux");
        factory.addCircuitBreakerCustomizer(circuitBreaker -> circuitBreaker.getEventPublisher()
            .onError(normalFluxErrorConsumer).onSuccess(normalFluxSuccessConsumer), "normalflux");
     };
}

1.5. 收集指标

加入依赖: org.springframework.boot:spring-boot-starter-actuator io.github.resilience4j:resilience4j-micrometer.

注意: 不需要额外的加入 micrometer-core ,因为已经存在于 spring-boot-starter-actuator

  • Sentinel
代码语言:javascript复制
<!-- SpringCloud依赖 -->
org.springframework.cloud:spring-cloud-starter-circuitbreaker-sentinal
  • Spring Retry

该项目为Spring应用程序提供了声明式重试支持。

环境:

JDK1.7 , MAVEN 3.0.5

代码语言:javascript复制
<!-- SpringCloud依赖 -->
org.springframework.cloud:spring-cloud-starter-circuitbreaker-spring-retry

Spring Retry通过使用CircuitBreakerRetryPolicy 和 有状态重试. 来提供断路器实现 。

1.1. 默认配置

代码语言:javascript复制
@Bean
public Customizer<SpringRetryCircuitBreakerFactory> defaultCustomizer() {
    return factory -> factory.configureDefault(id -> new SpringRetryConfigBuilder(id)
        .retryPolicy(new TimeoutRetryPolicy()).build());
}

1.2. 特殊配置

代码语言:javascript复制
@Bean
public Customizer<SpringRetryCircuitBreakerFactory> slowCustomizer() {
    return factory -> factory.configure(builder -> builder.retryPolicy(new SimpleRetryPolicy(1)).build(), "slow");
}
@Bean
public Customizer<SpringRetryCircuitBreakerFactory> slowCustomizer() {
    return factory -> factory.addRetryTemplateCustomizers(retryTemplate -> retryTemplate.registerListener(new RetryListener() {

        @Override
        public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
            return false;
        }
        @Override
        public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {

        }
        @Override
        public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {

        }
    }));
}

0 人点赞