兰若春夏

Resilience4j

2641 字 约 9 分钟

微服务容错简介

在高并发访问下，比如天猫双11，流量持续不断的涌入，服务之间的相互调用频率突然增加，引发系统负载过高，这时系统所依赖的服务的稳定性对系统的影响非常大，而且还有很多不确定因素引起雪崩，如网络连接中断，服务宕机等。一般微服务容错组件提供了限流、隔离、降级、熔断等手段，可以有效保护我们的微服务系统。

Resilience4j

https://resilience4j.readme.io/docs/getting-started

https://github.com/lmhmhl/Resilience4j-Guides-Chinese/blob/main/index.md

Resilience4j是一个轻量级容错框架，设计灵感来源于Netflix 的Hystrix框架，为函数式编程所设计。

Resilience4j 提供了一组高阶函数（装饰器），包括断路器，限流器，重试，隔离，可以对任何的函数式接口，lambda表达式，或方法的引用进行增强，并且这些装饰器可以进行叠加。这样做的好处是，你可以根据需要选择特定的装饰器进行组合。

1
2
3
4
5
6
7
8
9
Supplier<String> supplier = () -> service.sayHelloWorld(param1);

String result = Decorators.ofSupplier(supplier)
  .withBulkhead(Bulkhead.ofDefaults("name"))
  .withCircuitBreaker(CircuitBreaker.ofDefaults("name"))
  .withRetry(Retry.ofDefaults("name"))
  .withFallback(asList(CallNotPermittedException.class, BulkheadFullException.class),  
      throwable -> "Hello from fallback")
  .get()

在使用时，你不需要引入所有和Resilience4j相关的包，只需要引入所需要的即可。

核心模块

  • resilience4j-circuitbreaker: 熔断
  • resilience4j-ratelimiter: 限流
  • resilience4j-bulkhead: 隔离
  • resilience4j-retry: 自动重试
  • resilience4j-cache: 结果缓存
  • resilience4j-timelimiter: 超时处理

Resilience4j使用

添加依赖（按需引入）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
//使用注解时引入
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
    <version>1.7.1</version>
</dependency>

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-circuitbreaker</artifactId>
    <version>${resilience4j.version}</version>
</dependency>
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-ratelimiter</artifactId>
    <version>${resilience4j.version}</version>
</dependency>
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-retry</artifactId>
    <version>${resilience4j.version}</version>
</dependency>
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-bulkhead</artifactId>
    <version>${resilience4j.version}</version>
</dependency>

断路器（CircuitBreaker）

断路器通过有限状态机实现，有三个普通状态：关闭、开启、半开，还有两个特殊状态：禁用、强制开启。

enter image description here

断路器使用滑动窗口来存储和统计调用的结果。你可以选择基于调用数量的滑动窗口或者基于时间的滑动窗口。基于访问数量的滑动窗口统计了最近N次调用的返回结果。居于时间的滑动窗口统计了最近N秒的调用返回结果。

enter image description here

当熔断器关闭时，所有的请求都会通过熔断器。如果失败率超过设定的阈值，熔断器就会从关闭状态转换到打开状态，这时所有的请求都会被拒绝。当经过一段时间后，熔断器会从打开状态转换到半开状态，这时仅有一定数量的请求会被放入，并重新计算失败率，如果失败率超过阈值，则变为打开状态，如果失败率低于阈值，则变为关闭状态。

断路器使用滑动窗口来存储和统计调用的结果。你可以选择基于调用数量的滑动窗口或者基于时间的滑动窗口。基于访问数量的滑动窗口统计了最近N次调用的返回结果。居于时间的滑动窗口统计了最近N秒的调用返回结果。

除此以外，熔断器还会有两种特殊状态：DISABLED（始终允许访问）和**FORCED_OPEN（始终拒绝访问）。这两个状态不会生成熔断器事件（除状态装换外），并且不会记录事件的成功或者失败。退出这两个状态的唯一方法是触发状态转换或者重置熔断器。**

创建一个CircuitBreakerRegistry

Resilience4j使用基于ConcurrentHashMap的CircuitBreakerRegistry来保证线程安全和原子性，你可以使用CircuitBreakerRegistry管理（创建和检索）断路器实例，可以使用全局默认的CircuitBreakerConfig配置为所有的断路器实例创建一个CircuitBreakerRegistry。

1
2
CircuitBreakerRegistry circuitBreakerRegistry = 
  CircuitBreakerRegistry.ofDefaults();

创建和配置CircuitBreaker

你可以自定义CircuitBreakerConfig，为了创建自定义的CircuitBreakerConfig，你可以使用CircuitBreakerConfig建造器，你可以使用建造者模式来配置下面的属性。

配置属性 默认值 描述
failureRateThreshold 50 以百分比配置失败率阈值。当失败率等于或大于阈值时，断路器状态并关闭变为开启，并进行服务降级。
slowCallRateThreshold 100 以百分比的方式配置，断路器把调用时间大于slowCallDurationThreshold的调用视为满调用，当慢调用比例大于等于阈值时，断路器开启，并进行服务降级。
slowCallDurationThreshold 60000 [ms] 配置调用时间的阈值，高于该阈值的呼叫视为慢调用，并增加慢调用比例。
permittedNumberOfCallsInHalfOpenState 10 断路器在半开状态下允许通过的调用次数。
maxWaitDurationInHalfOpenState 0 断路器在半开状态下的最长等待时间，超过该配置值的话，断路器会从半开状态恢复为开启状态。配置是0时表示断路器会一直处于半开状态，直到所有允许通过的访问结束。
slidingWindowType COUNT_BASED 配置滑动窗口的类型，当断路器关闭时，将调用的结果记录在滑动窗口中。滑动窗口的类型可以是count-based或time-based。如果滑动窗口类型是COUNT_BASED，将会统计记录最近slidingWindowSize次调用的结果。如果是TIME_BASED，将会统计记录最近slidingWindowSize秒的调用结果。
slidingWindowSize 100 配置滑动窗口的大小。
minimumNumberOfCalls 100 断路器计算失败率或慢调用率之前所需的最小调用数（每个滑动窗口周期）。例如，如果minimumNumberOfCalls为10，则必须至少记录10个调用，然后才能计算失败率。如果只记录了9次调用，即使所有9次调用都失败，断路器也不会开启。
waitDurationInOpenState 60000 [ms] 断路器从开启过渡到半开应等待的时间。
automaticTransition FromOpenToHalfOpenEnabled false 如果设置为true，则意味着断路器将自动从开启状态过渡到半开状态，并且不需要调用来触发转换。创建一个线程来监视断路器的所有实例，以便在WaitDurationInOpenstate之后将它们转换为半开状态。但是，如果设置为false，则只有在发出调用时才会转换到半开，即使在waitDurationInOpenState之后也是如此。这里的优点是没有线程监视所有断路器的状态。
recordExceptions empty 记录为失败并因此增加失败率的异常列表。 除非通过ignoreExceptions显式忽略，否则与列表中某个匹配或继承的异常都将被视为失败。 如果指定异常列表，则所有其他异常均视为成功，除非它们被ignoreExceptions显式忽略。
ignoreExceptions empty 被忽略且既不算失败也不算成功的异常列表。 任何与列表之一匹配或继承的异常都不会被视为失败或成功，即使异常是recordExceptions的一部分。
recordException throwable -> true· By default all exceptions are recored as failures. 一个自定义断言，用于评估异常是否应记录为失败。 如果异常应计为失败，则断言必须返回true。如果出断言返回false，应算作成功，除非ignoreExceptions显式忽略异常。
ignoreException throwable -> false By default no exception is ignored. 自定义断言来判断一个异常是否应该被忽略，如果应忽略异常，则谓词必须返回true。 如果异常应算作失败，则断言必须返回false。 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 自定义断路器配置
CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
  .failureRateThreshold(50)
  .slowCallRateThreshold(50)
  .waitDurationInOpenState(Duration.ofMillis(1000))
  .slowCallDurationThreshold(Duration.ofSeconds(2))
  .permittedNumberOfCallsInHalfOpenState(3)
  .minimumNumberOfCalls(10)
  .slidingWindowType(SlidingWindowType.TIME_BASED)
  .slidingWindowSize(5)
  .recordException(e -> INTERNAL_SERVER_ERROR
                 .equals(getResponse().getStatus()))
  .recordExceptions(IOException.class, TimeoutException.class)
  .ignoreExceptions(BusinessException.class, OtherBusinessException.class)
  .build();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 熔断器半开时的缓冲区大小
resilience4j.circuitbreaker.configs.default.permitted-number-of-calls-in-half-open-state=3

# 熔断器关闭时的缓冲区大小
resilience4j.circuitbreaker.configs.default.sliding-window-size=10
# 熔断器从打开到半开需要的时间
resilience4j.circuitbreaker.configs.default.wait-duration-in-open-state=2s
# 熔断器打开的失败阈值
resilience4j.circuitbreaker.configs.default.failureRateThreshold=30

# 事件缓冲区大小
resilience4j.circuitbreaker.configs.default.eventConsumerBufferSize=10
# 健康监测
resilience4j.circuitbreaker.configs.default.registerHealthIndicator=true
# 是否自动从打开到半开，不需要触发
resilience4j.circuitbreaker.configs.default.automaticTransitionFromOpenToHalfOpenEnabled=true
# 记录的异常
resilience4j.circuitbreaker.configs.default.recordExceptions=cn.dragonsoftbravo.practice.exception.BusinessBException,cn.dragonsoftbravo.practice.exception.BusinessAException
# 忽略的异常
resilience4j.circuitbreaker.configs.default.ignoreExceptions=cn.dragonsoftbravo.practice.exception.BusinessAException


resilience4j.circuitbreaker.instances.backendA.failure-rate-threshold=50
#慢调用时间阈值，高于这个阈值的呼叫视为慢调用，并增加慢调用比例
resilience4j.circuitbreaker.instances.backendA.slow-call-duration-threshold=2s
#慢调用百分比阈值，断路器把调用时间大于slow-call-duration-threshold，视为慢调用
resilience4j.circuitbreaker.instances.backendA.slow-call-rate-threshold=30
resilience4j.circuitbreaker.instances.backendA.sliding-window-size=10
resilience4j.circuitbreaker.instances.backendA.sliding-window-type=TIME_BASED
resilience4j.circuitbreaker.instances.backendA.minimum-number-of-calls=2
resilience4j.circuitbreaker.instances.backendA.permitted-number-of-calls-in-half-open-state=2
#从OPEN到HALF_OPEN状态需要等待的时间
resilience4j.circuitbreaker.instances.backendA.wait-duration-in-open-state=2s

隔离(Builkhead)

Resilience4j提供了两种隔离的实现方式，可以限制并发执行的数量。

  • SemaphoreBulkhead使用了信号量
  • FixedThreadPoolBulkhead使用了有界队列和固定大小线程池

SemaphoreBulkhead使用了信号量，配置属性，如下表4-3所示。

配置属性 默认值 描述
maxConcurrentCalls 25 隔离允许线程并发执行的最大数量
maxWaitDuration 0 当达到并发调用数量时，新的线程执行时将被阻塞，这个属性表示最长的等待时间。 
1
2
3
4
5
//为Bulkhead创建自定义的配置
BulkheadConfig config = BulkheadConfig.custom()
    .maxConcurrentCalls(150)
    .maxWaitDuration(Duration.ofMillis(500))
    .build();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#隔离允许并发线程执行的最大数量
resilience4j.bulkhead.configs.default.max-concurrent-calls=5 
#当达到并发调用数量时，新的线程的阻塞时间
resilience4j.bulkhead.configs.default.max-wait-duration=20ms
resilience4j.bulkhead.instances.backendA.base-config=default
resilience4j.bulkhead.instances.backendB.max-concurrent-calls=20
resilience4j.bulkhead.instances.backendB.max-wait-duration.=10ms
#最大线程池大小
resilience4j.thread-pool-bulkhead.configs.default.max-thread-pool-size=4
#核心线程池大小
resilience4j.thread-pool-bulkhead.configs.default.core-thread-pool-size=2
#队列容量
resilience4j.thread-pool-bulkhead.configs.default.queue-capacity=2 

resilience4j.thread-pool-bulkhead.instances.backendA.base-config=default
resilience4j.thread-pool-bulkhead.instances.backendB.max-thread-pool-size=1
resilience4j.thread-pool-bulkhead.instances.backendB.core-thread-pool-size=1
resilience4j.thread-pool-bulkhead.instances.backendB.queue-capacity=1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/**
 * 信号量隔离
 */
BulkheadConfig config = BulkheadConfig.custom()
        .maxConcurrentCalls(5)
        .maxWaitDuration(Duration.ofMillis(500))
        .build();


/**
 * 线程池隔离
 */
ThreadPoolBulkheadConfig  poolBulkheadConfig= ThreadPoolBulkheadConfig.custom()
        .queueCapacity(2)
        .maxThreadPoolSize(4)
        .coreThreadPoolSize(2)
        .build();

Bulkhead bulkhead = Bulkhead.of("name", config);

ThreadPoolBulkhead threadPoolBulkhead = ThreadPoolBulkhead.of("name", poolBulkheadConfig);

限流(RateLimiter)

Resilience4j提供了一个限流器，它将从epoch开始的所有纳秒划分为多个周期。每个周期的持续时间RateLimiterConfig.limitRefreshPeriod。在每个周期开始时，限流器将活动权限数设置为RateLimiterConfig.limitForPeriod。期间， 对于限流器的调用者，它看起来确实是这样的，但是对于AtomicRateLimiter实现，如果RateLimiter未被经常使用，则会在后台进行一些优化，这些优化将跳过此刷新。

限流器的默认实现是AtomicRateLimiter，它通过原子引用管理其状态。这个AtomicRateLimiter状态完全不可变，并且具有以下字段：

  • activeCycle -上次调用的周期号
  • activePermissions -在上次调用结束后，可用的活跃权限数。如果保留了某些权限，则可以为负。
  • nanosToWait - 最后一次调用要等待的纳秒数

还有一个使用信号量的SemaphoreBasedRateLimiter和一个调度程序，它将在每个RateLimiterConfig#limitRefreshPeriod之后刷新活动权限数。

Resilience4j的限流模块RateLimter基于滑动窗口，和令牌桶限流算法，配置如下，如下表4-5所示。

属性 默认值 描述
timeoutDuration 5秒 线程等待权限的默认等待时间
limitRefreshPeriod 500纳秒 限流器每隔limitRefreshPeriod刷新一次，将允许处理的最大请求数量重置为limitForPeriod。
limitForPeriod 50 在一次刷新周期内，允许执行的最大请求数 
1
2
3
4
5
6
7
8
9
10
11
12
#线程等待权限的默认等待时间
resilience4j.ratelimiter.configs.default.timeout-duration=5
#限流器每隔1s刷新一次，将允许处理的最大请求重置
resilience4j.ratelimiter.configs.default.limit-refresh-period=1s
#在一个刷新周期内，允许执行的最大请求数
resilience4j.ratelimiter.configs.default.limit-for-period=2

resilience4j.ratelimiter.instances.backendA.base-config=default

resilience4j.ratelimiter.instances.backendB.timeout-duration=5
resilience4j.ratelimiter.instances.backendB.limit-refresh-period=1s
resilience4j.ratelimiter.instances.backendB.limit-for-period=5
1
2
3
4
5
6
7
RateLimiterConfig limiterConfig = RateLimiterConfig.custom()
        .limitRefreshPeriod(Duration.ofSeconds(1))
        .limitForPeriod(10)
        .timeoutDuration(Duration.ofMillis(25))
        .build();

RateLimiter rateLimiter = RateLimiterRegistry.of(limiterConfig).rateLimiter("name2");

重试(Retry)

创建 RetryRegistry

就像断路器模块一样，这么模块提供了在内存中的RetryRegistry，你可以使用这个管理（创建和检索）Retry实例。

1
RetryRegistry retryRegistry = RetryRegistry.ofDefaults();
创建和配置重试

你可以提供一个自定义的全局RetryConfig，为了创建一个自定义的全局RetryConfig，可以使用建造者模式对RetryConfig进行配置。

  • 最大的重试次数 。
  • 连续两次重试之间的时间间隔。
  • 自定义断言机制，评估一个响应是否可以触发重试机制。
  • 自定义断言机制，评估一个异常是否可以出发重试机制。
  • 定义一个异常的列表，这些异常能够触发重试机制。
  • 定义一个异常的列表，这些异常应该被忽略并且不会触发重试机制。
属性 默认值 描述
maxAttempts 3 最大重试次数
waitDuration 500 [ms] 两次重试之间的时间间隔
intervalFunction numOfAttempts -> waitDuration 修改重试间隔的函数。默认情况下，等待时间保持不变。
retryOnResultPredicate result -> false 配置用于计算是否应重试的断言。如果要重试，断言必须返回true，否则返回false。
retryOnExceptionPredicate throwable -> true 配置一个断言，判断某个异常发生时，是否要进行重试。如果要重试，断言必须返回true，否则必须返回false。
retryExceptions empty 配置一个Throwable类型的列表，被记录为失败类型，需要进行重试，支持子类型。
ignoreExceptions empty 配置一个Throwable类型的列表，被记录为忽略类型，不会进行重试，支持子类型。 
1
2
3
4
5
resilience4j.retry.configs.default.max-attempts=3
resilience4j.retry.configs.default.wait-duration=2s
resilience4j.retry.configs.default.enable-exponential-backoff=true
resilience4j.retry.configs.default.ignore-exceptions=java.lang.IllegalStateException,com.beust.jcommander.ParameterException
resilience4j.retry.configs.default.retry-exceptions=java.util.concurrent.TimeoutException,java.net.ConnectException
1
2
3
4
5
6
7
8
RetryConfig config = RetryConfig.custom()
  .maxAttempts(2)
  .waitDuration(Duration.ofMillis(1000))
  .retryOnResult(response -> response.getStatus() == 500)
  .retryOnException(e -> e instanceof WebServiceException)
  .retryExceptions(IOException.class, TimeoutException.class)
  .ignoreExceptions(BusinessException.class, OtherBusinessException.class)
  .build();