自从用上了resilience4j，系统的容错性终于强壮起来了
1 前言
Resilience4j是一个轻量级、高性能并易于使用容错框架，设计灵感来源于Netflix 的Hystrix框架，采用广受欢迎的函数式编程风格。而且除了vavr，没有其他的外部依赖。它提供了断路器、限流器、重试、隔离、超时等容错设计时常用工具，这些工具可以灵活选择、组合使用，以使系统具备更好的容错功能
需要使用resilience4j不同部分的功能，只需要在pom文件中引入相应组件的依赖，也可以通过以下方式引入resilience4j的所有组件：
spring boot环境：
<dependency>
<groupId>io.github.resilience4j</groupId>
<artifactId>resilience4j-spring-boot3</artifactId>
</dependency>
spring cloud环境：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-circuitbreaker-resilience4j</artifactId>
</dependency>
2 circuitbreaker
circuitbreaker有三个典型的状态：CLOSED、OPEN、HALF_OPEN，以及两个特殊的状态：DISABLED和FORCED_OPEN。当他处于CLOSED状态时，正如电气中的断路器合上一样，所有的调用可以正常执行。在这个状态，circuitbreaker会持续统计调用的结果情况，如果失败率或慢调用率超过了指定的阈值，circuitbreaker的状态会变为OPEN状态。也如电气中的断路器一样，OPEN状态的circuitbreaker不允许任何调用，所有的调用都会返回CallNotPermittedException异常。经过设定的一段时间后，circuitbreaker的状态会变为HALF_OPEN状态。在这个状态中circuitbreaker允许部分的调用，同时持续统计调用的结果情况，如果失败率或慢调用率仍然低于指定的阈值，则仍然回到OPEN状态，如果失败率和慢调用率都高于指定的阈值，则变为CLOSED状态，恢复正常调用。DISABLED和FORCED_OPEN则是两个管理状态，这两个状态下，上述的断路器逻辑不生效，DISABLED强制允许所有的调用，FORCED_OPEN强制拒绝所有的调用。
circuitbreaker通过滑动窗口来统计调用的失败率和慢调用率。滑动窗口分为基于计数的滑动窗口和基于时间的滑动窗口。滑动窗口在实现上其实就是一个ring-buffer，基于计数的滑动窗口记录了最近N次调用的情况，然后根据这N次调用的失败率和慢调用率确定circuitbreaker的状态。基于时间的滑动窗口每间隔一个滑动步长时间窗口向前推进，每一个滑动步长时间区间维护一个桶，在这个时间区间内的调用结果统计到该桶中，然后根据整个窗口中所有桶内的统计数据综合得出调用的失败率和慢调用率，然后确定circuitbreaker的状态。
默认情况下，circuitbreaker把所有抛出异常的调用都统计为失败的调用，用户也可以指定具体哪些异常才统计为失败调用。
需要注意的是，circuitbreaker保护的对象并不是自己，而是被调用的对象。因为在微服务环境中，往往会存在重试机制，当被调用对象无法正常处理请求时，不使用circuitbreaker机制的话，被调用对象的流量压力会更高，从而更容易崩溃，进而导致整个系统崩溃。circuitbreaker就是在检测到被调用对象不正常时，暂停对它的调用，从而保护被调用对象，进而保护整个系统。
2.1 创建circuitbreaker
我们一般通过CircuitBreakerRegistry来统一管理我们的circuitbreaker实例。可以通过以下方式创建一个CircuitBreakerRegistry实例：
CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults();
这样创建的CircuitBreakerRegistry的默认断路器配置是系统默认的，如果想自己设置默认配置，可以创建一个自己的CircuitBreakerConfig，然后如下创建CircuitBreakerRegistry实例：
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();
CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.of(circuitBreakerConfig);
通过CircuitBreakerRegistry我们可以新建或获取已存在的circuitbreaker实例：
// 返回一个名字为name1的circuitbreaker，如果registry中已经存在，直接获取；如果还不存在，则创建一个新的;circuitbreaker的配置为默认配置
CircuitBreaker circuitBreakerWithDefaultConfig = circuitBreakerRegistry.circuitBreaker("name1");
上面创建的circuitbreaker实例的配置是registry默认的配置，可以通过下面方式指定circuitbreaker的配置：
CircuitBreaker circuitBreakerWithDefaultConfig = circuitBreakerRegistry.circuitBreaker("name1", circuitBreakerConfig);
或者：
CircuitBreakerConfig defaultConfig = circuitBreakerRegistry
.getDefaultConfig();
CircuitBreakerConfig overwrittenConfig = CircuitBreakerConfig
.from(defaultConfig)
.waitDurationInOpenState(Duration.ofSeconds(20))
.build();
// 先将配置添加到registry中
circuitBreakerRegistry.addConfiguration("name1Config", overwrittenConfig);
CircuitBreaker circuitbreaker = circuitBreakerRegistry.circuitBreaker("name1", "name1Config");
当然，你也可以选择不通过circuitBreakerRegistry管理，而是直接创建circuitbreaker，只是这种情况不常见。
CircuitBreaker customCircuitBreaker = CircuitBreaker.of("testName", circuitBreakerConfig);
circuitbreaker的详细配置信息如下所示：
| 配置属性 | 默认值 | 描述 |
|---|---|---|
| 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] | 断路器从开启过渡到半开应等待的时间。 |
| automaticTransitionFromOpenToHalfOpenEnabled | 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 -> falseBy default no exception is ignored. | 自定义断言来判断一个异常是否应该被忽略，如果应忽略异常，则谓词必须返回true。如果异常应算作失败，则断言必须返回false。 |
2.2 应用circuitbreaker
我们可以通过circuitbreaker实例装饰、执行被调用方法：
Object result = circuitbreaker.executeSupplier(()-> { return service.doSomething() ; });
circuitbreaker对Callable、Supplier、Runnable、Consumer、CheckedRunable、CheckedSupplier、CheckedConsumer或CompletionStage接口都提供了相应的执行方法:
executeCallable()
executeCheckedRunnable()
executeCheckedSupplier()
executeCompletionStage()
executeRunnable()
executeSupplier()
我们也可以将装饰和执行分开，这样性能会高些：
Supplier supplier = circuitbreaker.decorateSupplier(()-> { return service.doSomething() ; });
Object result = supplier.get();
同样，circuitbreaker对Callable、Supplier、Runnable、Consumer、CheckedRunable、CheckedSupplier、CheckedConsumer或CompletionStage接口都提供了相应的装饰方法:
decorateCallable()
decorateCheckedRunnable()
decorateCheckedSupplier()
decorateCompletionStage()
decorateRunnable()
decorateSupplier()
decorateConsumer()
decorateFuture()
2.3 监听circuitbreaker事件
监听CircuitBreakerRegistry的注册和注销事件：
CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults();
circuitBreakerRegistry.getEventPublisher()
.onEntryAdded(entryAddedEvent -> {
// 注册了新的circuitbreaker
CircuitBreaker addedCircuitBreaker = entryAddedEvent.getAddedEntry();
LOG.info("CircuitBreaker {} added", addedCircuitBreaker.getName());
})
.onEntryRemoved(entryRemovedEvent -> {
// 注销circuitbreaker
CircuitBreaker removedCircuitBreaker = entryRemovedEvent.getRemovedEntry();
LOG.info("CircuitBreaker {} removed", removedCircuitBreaker.getName());
});
监听circuitbreaker的状态转换和调用结果等事件：
// 分别处理不同的事件
circuitBreaker.getEventPublisher()
.onSuccess(event -> logger.info(...))
.onError(event -> logger.info(...))
.onIgnoredError(event -> logger.info(...))
.onReset(event -> logger.info(...))
.onStateTransition(event -> logger.info(...));
// 统一处理所有的事件
circuitBreaker.getEventPublisher()
.onEvent(event -> logger.info(...));
resilience4j限流器、隔离、超时、重试的registry、config、执行、事件监听机制与circuitbreaker一致，后续不一一介绍，主要如何使用
3 ratelimiter
ratelimiter可以限制单位时间内可以调用的次数，以避免系统过载，引发故障，甚至造成系统崩溃。
实现ratelimiter一般有以下思路：
- 滚动时间窗口
在指定的时间窗口内最多调用指定的次数。窗口每次移动窗口大小的时间，相邻窗口时间不重叠。这种方式调用速率可能不均衡，可能出现突发峰值速率。 2. 滑动时间窗口
与滚动时间窗口相同，在指定的时间窗口内最多调用指定的次数。但窗口每次只移动指定步长（步长小于窗口大小，比如十分之一的窗口大小）的时间，相邻窗口时间重叠。 3. 令牌桶
令牌桶会以固定的速率往令牌桶中放入令牌，令牌桶中的令牌数有一个最大值，任何时候令牌桶中的令牌数不会超过最大值。调用前需要先到令牌桶中获取相应数量的令牌，只有令牌桶中具有相应的令牌时才运行调用 4. 漏桶
漏桶像一个装有水或沙子的漏斗，调用像漏桶中的水或沙子一样按固定的速率流出
resilience4j的ratelimiter有两种实现，一种基于SemaphoreBasedRateLimiter（令牌桶），另一种基于AtomicRateLimiter，AtomicRateLimiter针对有些时间内没有调用的场景进行了优化。
我们可以通过以下方式创建一个ratelimiter：
// 自定义配置
RateLimiterConfig config = RateLimiterConfig.custom()
.limitRefreshPeriod(Duration.ofSeconds(1))
.limitForPeriod(10)
.timeoutDuration(Duration.ofMillis(25))
.build();
// 创建Registry
RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(config);
// 使用默认配置创建
RateLimiter rateLimiterWithDefaultConfig = rateLimiterRegistry
.rateLimiter("name1");
// 使用自定义配置创建
RateLimiter rateLimiterWithCustomConfig = rateLimiterRegistry
.rateLimiter("name2", config);
ratelimiter的配置参数说明：
| 属性 | 默认值 | 描述 |
|---|---|---|
| timeoutDuration | 5秒 | 线程等待调用许可（如可用令牌）的等待时间 |
| limitRefreshPeriod | 500纳秒 | 限流器每隔limitRefreshPeriod刷新一次，将允许处理的最大请求数量重置为limitForPeriod。 |
| limitForPeriod | 50 | 在一次刷新周期内，允许执行的最大请求数 |
4 bulkhead
bulkhead实质上是限制调用的并发度。它有两种实现：SemaphoreBulkhead和FixedThreadPoolBulkhead。SemaphoreBulkhead通过semaphore数量控制并发数，而不限制调用在哪个线程池中执行。FixedThreadPoolBulkhead则通过创建指定大小的线程池来控制并发数，调用发生在同一个线程池中。
bulkhead配置参数：
| 配置属性 | 默认值 | 描述 |
|---|---|---|
| maxConcurrentCalls | 25 | 允许并发执行的最大数量 |
| maxWaitDuration | 0 | 当达到并发调用数量时，新的调用将被阻塞，这个属性表示最长的等待时间。 |
创建SemaphoreBulkhead：
//为Bulkhead创建自定义的配置
BulkheadConfig config = BulkheadConfig.custom()
.maxConcurrentCalls(150)
.maxWaitDuration(Duration.ofMillis(500))
.build();
// 使用自定义全局配置创建BulkheadRegistry
BulkheadRegistry registry = BulkheadRegistry.of(config);
// 使用默认的配置从registry中创建Bulkhead
Bulkhead bulkheadWithDefaultConfig = registry.bulkhead("name1");
// 使用自定义的配置从regidtry中创建bulkhead
Bulkhead bulkheadWithCustomConfig = registry.bulkhead("name2", custom);
创建FixedThreadPoolBulkhead：
ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
.maxThreadPoolSize(10)
.coreThreadPoolSize(2)
.queueCapacity(20)
.build();
ThreadPoolBulkhead bulkhead = ThreadPoolBulkhead.of("name", config);
CompletionStage<String> supplier = ThreadPoolBulkhead
.executeSupplier(bulkhead, backendService::doSomething);
5 retry
retry操作在容错性设计中太常见了，resilience4j也提供了相应的工具。
创建retry：
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();
// 使用自定义的配置创建RetryRegistry
RetryRegistry registry = RetryRegistry.of(config);
// 使用默认的配置从Registry中获取和创建一个Retry
Retry retryWithDefaultConfig = registry.retry("name1");
// 使用自定义的配置从Registry中获取和创建一个Retry
RetryConfig custom = RetryConfig.custom()
.waitDuration(Duration.ofMillis(100))
.build();
Retry retryWithCustomConfig = registry.retry("name2", custom);
retry配置属性说明：
| 属性 | 默认值 | 描述 |
|---|---|---|
| maxAttempts | 3 | 最大重试次数 |
| waitDuration | 500 [ms] | 两次重试之间的时间间隔 |
| intervalFunction | numOfAttempts -> waitDuration | 修改重试间隔的函数。默认情况下，等待时间保持不变。 |
| retryOnResultPredicate | result -> false | 配置用于计算是否应重试的断言。如果要重试，断言必须返回true，否则返回false。 |
| retryOnExceptionPredicate | throwable -> true | 配置一个断言，判断某个异常发生时，是否要进行重试。如果要重试，断言必须返回true，否则必须返回false。 |
| retryExceptions | empty | 配置一个Throwable类型的列表，被记录为失败类型，需要进行重试，支持子类型。 |
| ignoreExceptions | empty | 配置一个Throwable类型的列表，被记录为忽略类型，不会进行重试，支持子类型。 |
6 timelimiter
超时控制在容错性设计中与重试一样常见，resilience4j同样提供了相应的工具
创建timelimiter实例：
TimeLimiterConfig config = TimeLimiterConfig.custom()
// 是否停止正在异步执行的调用
.cancelRunningFuture(true)
// 超时时间
.timeoutDuration(Duration.ofMillis(500))
.build();
// 使用自定义的全局配置创建一个TimeLimiterRegistry
TimeLimiterRegistry timeLimiterRegistry = TimeLimiterRegistry.of(config);
//registry使用默认的配置创建一个TimeLimiter
TimeLimiter timeLimiterWithDefaultConfig = registry.timeLimiter("name1");
// 使用自定义的配置创建一个TimeLimiter实例
TimeLimiterConfig config = TimeLimiterConfig.custom()
.cancelRunningFuture(false)
.timeoutDuration(Duration.ofMillis(1000))
.build();
TimeLimiter timeLimiterWithCustomConfig = registry.timeLimiter("name2", config);
timelimiter与其他的组件有一个区别，就是它的调用必须放在与timelimiter不同的线程中执行，否则不生效。示例代码如下：
// 被调用的服务
HelloWorldService helloWorldService = mock(HelloWorldService.class);
// 创建一个限时器实例
TimeLimiter timeLimiter = TimeLimiter.of(Duration.ofSeconds(1));
// 被调用服务的运行的线程池
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
// 返回CompletableFuture类型的非阻塞变量
CompletableFuture<String> result = timeLimiter.executeCompletionStage(
scheduler, () -> CompletableFuture.supplyAsync(helloWorldService::sayHelloWorld)).toCompletableFuture();
// 阻塞方式，实际上是调用了future.get(timeoutDuration, MILLISECONDS)
String result = timeLimiter.executeFutureSupplier(
() -> CompletableFuture.supplyAsync(() -> helloWorldService::sayHelloWorld));
7 通过注解使用resilience4j
上面介绍的方式都是通过代码创建相应实例然后调用服务的方式，在spring boot/cloud环境中可以通过注解的方式添加resilience4j的功能。
使用注解的方式，确保项目添加了AOP依赖，否则不生效：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
然后在配置文件中配置resilience4j的组件(就是上面我们介绍的配置参数)：
resilience4j.circuitbreaker:
configs:
default:
slidingWindowSize: 100
permittedNumberOfCallsInHalfOpenState: 10
waitDurationInOpenState: 10000
failureRateThreshold: 60
eventConsumerBufferSize: 10
registerHealthIndicator: true
someShared:
slidingWindowSize: 50
permittedNumberOfCallsInHalfOpenState: 10
instances:
instances:
backendA:
baseConfig: default
waitDurationInOpenState: 5000
backendB:
baseConfig: someShared
registerHealthIndicator: true
slidingWindowSize: 10
permittedNumberOfCallsInHalfOpenState: 3
slidingWindowType: TIME_BASED
minimumNumberOfCalls: 20
waitDurationInOpenState: 50s
failureRateThreshold: 50
eventConsumerBufferSize: 10
recordFailurePredicate: io.github.robwin.exception.RecordFailurePredicate
resilience4j.retry:
instances:
backendA:
maxRetryAttempts: 3
waitDuration: 10s
enableExponentialBackoff: true
exponentialBackoffMultiplier: 2
retryExceptions:
- org.springframework.web.client.HttpServerErrorException
- java.io.IOException
ignoreExceptions:
- io.github.robwin.exception.BusinessException
backendB:
maxRetryAttempts: 3
waitDuration: 10s
retryExceptions:
- org.springframework.web.client.HttpServerErrorException
- java.io.IOException
ignoreExceptions:
- io.github.robwin.exception.BusinessException
resilience4j.bulkhead:
instances:
backendA:
maxConcurrentCalls: 10
backendB:
maxWaitDuration: 10ms
maxConcurrentCalls: 20
resilience4j.thread-pool-bulkhead:
instances:
backendC:
maxThreadPoolSize: 1
coreThreadPoolSize: 1
queueCapacity: 1
resilience4j.ratelimiter:
instances:
backendA:
limitForPeriod: 10
limitRefreshPeriod: 1s
timeoutDuration: 0
registerHealthIndicator: true
eventConsumerBufferSize: 100
backendB:
limitForPeriod: 6
limitRefreshPeriod: 500ms
timeoutDuration: 3s
resilience4j.timelimiter:
instances:
backendA:
timeoutDuration: 2s
cancelRunningFuture: true
backendB:
timeoutDuration: 1s
cancelRunningFuture: false
然后就可以在需要保护的接口上添加相应的组件了
@RestController
@RequestMapping(value = "/")
public class Controller {
@CircuitBreaker(name = "backendB", fallbackMethod = "fallback")
@RateLimiter(name = "backendB")
@Bulkhead(name = "backendB")
@Retry(name = "backendB", fallbackMethod = "fallback")
@TimeLimiter(name = "backendB")
@GetMapping(value = "/test")
public String test() {
return "hello";
}
private String fallback(IllegalArgumentException e) {
return "hello fallback";
}
}
需要注意的是，通过注解的方式默认执行的顺序如下，bulkhead先执行，最后是retry：
Retry ( CircuitBreaker ( RateLimiter ( TimeLimiter ( Bulkhead ( CalledFunction ) ) ) ) )
然而，你可以通过如下配置修改执行的顺序：
resilience4j:
circuitbreaker:
circuitBreakerAspectOrder: 1
retry:
retryAspectOrder: 2