Soring Cloud -- Resilience4j简介

参考: GitHub - resilience4j/resilience4j: Resilience4j is a fault tolerance library designed for Java8 and functional programming

 Resilience4j_zhuhuan_Life的博客-CSDN博客_resilience4j

Central Repository: io/github/resilience4j

官方文档

github

SpringCloud之Resilience4J

https://www.jianshu.com/p/5527f5c75562

一、简介

        Netflix的Hystri停更后，SpringCloud家族推荐Resilience4j。在国内阿里也推出了Sentinel。

        随着微服务的流行，熔断作为其中一项很重要的技术也广为人知。当微服务的运行质量低于某个临界值时，启动熔断机制，暂停微服务调用一段时间，以保障后端的微服务不会因为持续过负荷而宕机。

        Hystrix官方已停止维护，官方推荐使用Resilience4j来替代Hystrix实现服务治理。作为新一代的熔断器，Resilience4j有很多优势，比如依赖少，模块化程度较好等优势。

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

• Resilience4j是一个轻量级、易于使用的容错库，其灵感来自Netflix Hystrix，但专为Java 8和函数式编程设计。
• Resilience4j提供高阶函数（decorators）来增强任何功能接口、lambda表达式或方法引用，包括断路器、速率限制器、重试或舱壁。可以在任何函数接口、lambda表达式或方法引用上使用多个装饰器。
• circuitbreaker组件实现了断路器功能，是基于内存的断路器，采用ConcurrentHashMap来实现。

二、核心组件

要使用Resilience4j，不需要引入所有依赖，只需要选择你需要的，Resilience4j提供了以下的核心模块和拓展模块：

三、 优缺点分析 ：

 系统自适应:  结合应用的机器负载、CPU 使用率，整体平均响应时间、入口 QPS 和并发线程数等几个维度的监控指标从而决定是否调用进行限流操做

四、并发控制器(舱壁，Bulkhead)

Bulkhead（舱壁）是用来控制并行(parallel)调用的次数。Resilience4j提供了两种舱壁模式的实现，可用于限制并发执行的次数：

1. SemaphoreBulkhead（信号量舱壁，默认），基于Java并发库中的Semaphore实现。
2. FixedThreadPoolBulkhead（固定线程池舱壁），它使用一个有界队列和一个固定线程池。

由于基于信号量的Bulkhead能很好地在多线程和I/O模型下工作，所以选择介绍基于信号量的Bulkhead的使用。

4.1 可配置参数

4.2 测试demo

1.首先添加POM依赖。不需要引入新的依赖，已经集成在resilience4j-spring-boot2中。

<dependency>
    <groupId>io.github.resilience4jgroupId>
    <artifactId>resilience4j-spring-boot2artifactId>
    <version>1.2.0version>
dependency>

2.并发控制器配置（application.yml）

package com.lx.cloud.demo.controller;
 
import com.alibaba.fastjson.JSONObject;
import com.lx.cloud.demo.entity.User;
import io.github.resilience4j.bulkhead.BulkheadFullException;
import io.github.resilience4j.bulkhead.BulkheadRegistry;
import io.github.resilience4j.bulkhead.annotation.Bulkhead;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
/**
 * Resilience4jController
 * @author lx
 */
@RestController
@Slf4j
public class Resilience4jController {
    @Autowired
    private BulkheadRegistry bulkheadRegistry;
 
    /**
     * 并发
     * @return
     * @throws Exception
     */
    @GetMapping("/bulkheadTestNoFallbackMethod")
    @Bulkhead(name = "bulkheadA")
    public User bulkheadTestNoFallbackMethod() throws Exception{
        Thread.sleep(500);
        User user = new User("liuxiao", 27);
        System.out.println(JSONObject.toJSONString(user));
        return user;
    }
 
    /**
     * 并发降级
     * @return
     * @throws Exception
     */
    @GetMapping("/bulkheadTestFallbackMethod")
    @Bulkhead(name = "bulkheadA",fallbackMethod = "fallBackByBulkhead")
    public User bulkheadTestFallbackMethod() throws Exception{
        Thread.sleep(5000);
        User user = new User("liuxiao", 27);
        System.out.println(JSONObject.toJSONString(user));
        return user;
    }
 
    //降级处理
    private User fallBackByBulkhead(BulkheadFullException e){
        System.out.println("并发控制器已经打开，拒绝访问被保护方法~");
        io.github.resilience4j.bulkhead.Bulkhead bulkheadA = bulkheadRegistry.bulkhead("bulkheadA");
        io.github.resilience4j.bulkhead.Bulkhead.Metrics metrics = bulkheadA.getMetrics();
        System.out.println("方法降级中："  + "metrics[ availableConcurrentCalls=" + metrics.getAvailableConcurrentCalls() +
                ", maxAllowedConcurrentCalls=" + metrics.getMaxAllowedConcurrentCalls() +
                " ]"
        );
        User user = new User("并发控制器降级测试", 0);
        System.out.println(JSONObject.toJSONString(user));
        return user;
    }
}

 使用Jmeter进行并发测试，调用bulkheadTestFallbackMethod服务降级方法。同时发送10个请求。控制台输出结果如下：

并发控制器已经打开，拒绝访问被保护方法~
方法降级中：metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=5 ]
{"age":0,"name":"并发控制器降级测试"}
并发控制器已经打开，拒绝访问被保护方法~
方法降级中：metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=5 ]
{"age":0,"name":"并发控制器降级测试"}
并发控制器已经打开，拒绝访问被保护方法~
方法降级中：metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=5 ]
{"age":0,"name":"并发控制器降级测试"}
并发控制器已经打开，拒绝访问被保护方法~
方法降级中：metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=5 ]
{"age":0,"name":"并发控制器降级测试"}
并发控制器已经打开，拒绝访问被保护方法~
方法降级中：metrics[ availableConcurrentCalls=0, maxAllowedConcurrentCalls=5 ]
{"age":0,"name":"并发控制器降级测试"}
{"age":27,"name":"liuxiao"}
{"age":27,"name":"liuxiao"}
{"age":27,"name":"liuxiao"}
{"age":27,"name":"liuxiao"}
{"age":27,"name":"liuxiao"}

五、信号量舱壁（SemaphoreBulkhead） 

当信号量存在剩余时进入系统的请求会直接获取信号量并开始业务处理。当信号量全被占用时，接下来的请求将会进入阻塞状态，SemaphoreBulkhead提供了一个阻塞计时器，如果阻塞状态的请求在阻塞计时内无法获取到信号量则系统会拒绝这些请求。若请求在阻塞计时内获取到了信号量，那将直接获取信号量并执行相应的业务处理。

六、固定线程池舱壁（FixedThreadPoolBulkhead） 

FixedThreadPoolBulkhead的功能与SemaphoreBulkhead一样也是用于限制并发执行的次数的，但是二者的实现原理存在差别而且表现效果也存在细微的差别。FixedThreadPoolBulkhead使用一个固定线程池和一个等待队列来实现舱壁。当线程池中存在空闲时，则此时进入系统的请求将直接进入线程池开启新线程或使用空闲线程来处理请求。当线程池无空闲时接下来的请求将进入等待队列，若等待队列仍然无剩余空间时接下来的请求将直接被拒绝。在队列中的请求等待线程池出现空闲时，将进入线程池进行业务处理。

可以看到FixedThreadPoolBulkhead和SemaphoreBulkhead一个明显的差别是FixedThreadPoolBulkhead没有阻塞的概念，而SemaphoreBulkhead没有一个队列容量的限制。

七、限流器（Rate Limiter）

       限速器（Rate Limiter）的功能是防止突然的过量请求导致系统不堪重负，RateLimiter使用一个刷新周期的概念，限定在一个固定刷新周期内可处理的最大请求数量。若在某一个刷新周期内的请求数量已经达到最大，则本周期内接下来的请求将进入阻塞状态，如果在最大阻塞计时内新的刷新周期开启，则阻塞状态的请求将进入新的周期内进行处理。如最大的阻塞计时内新的刷新周期并未开启，则此时超出阻塞计时的那些请求将被直接拒绝。

       高频控制可以限制服务调用频率，Resilience4j的RateLimiter可以对频率进行纳秒级别的控制，在每一个周期刷新可以调用的次数，还可以设定线程等待权限的时间，一般用于服务提供方，保护自己不受到冲击。

7.1 配置参数

7.2 测试demo

1.首先添加POM依赖(已经集成在resilience4j-spring-boot2中)

<dependency>
    <groupId>io.github.resilience4jgroupId>
    <artifactId>resilience4j-spring-boot2artifactId>
    <version>1.2.0version>
dependency>

2.限流控制器配置（application.yml）

resilience4j:
  ratelimiter:
    configs:
      default:
        limitForPeriod: 5  # 一个限制周期内可访问次数
        limitRefreshPeriod: 10000 # 限制周期,每个周期之后，速率限制器将重置回limitForPeriod值
        timeoutDuration: 5000 # 线程等待允许执行时间
    instances:
      ratelimiterA: # 限流器的名字
        baseConfig: default
        limitForPeriod: 3 # 一个限制周期内可访问次数
      ratelimiterB: # 限流器的名字
        baseConfig: default
        limitRefreshPeriod: 10s

3.使用注解的方式限流控制器 

package com.lx.cloud.demo.controller;
 
import com.alibaba.fastjson.JSONObject;
import com.lx.cloud.demo.entity.User;
import io.github.resilience4j.ratelimiter.RateLimiterRegistry;
import io.github.resilience4j.ratelimiter.RequestNotPermitted;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.util.Date;
import java.util.concurrent.TimeoutException;
 
/**
 * Resilience4jController
 * @author lx
 */
@RestController
@Slf4j
public class Resilience4jController {
 
    @Autowired
    private RateLimiterRegistry rateLimiterRegistry;
 
    /**
     * 限流
     * @return
     * @throws Exception
     */
    @GetMapping("/ratelimiterTestNoFallbackMethod")
    @RateLimiter(name = "ratelimiterA")
    public User ratelimiterTestNoFallbackMethod() throws Exception{
        User user = new User("liuxiao", 27);
        log.info(JSONObject.toJSONString(user) + new Date());
        return user;
    }
 
    /**
     * 限流有fallback方法
     * @return
     * @throws Exception
     */
    @GetMapping("/ratelimiterTestFallbackMethod")
    @RateLimiter(name = "ratelimiterA",fallbackMethod = "fallBackByRatelimiter")
    public User ratelimiterTestFallbackMethod() throws TimeoutException, InterruptedException {
        User user = new User("liuxiao", 27);
        log.info(JSONObject.toJSONString(user));
        return user;
    }
 
    private User fallBackByRatelimiter(RequestNotPermitted e){
        log.info("限流控制器已经打开，拒绝访问被保护方法~");
        io.github.resilience4j.ratelimiter.RateLimiter ratelimiterA = rateLimiterRegistry.rateLimiter("ratelimiterA");
        io.github.resilience4j.ratelimiter.RateLimiter.Metrics metrics = ratelimiterA.getMetrics();
        log.info("方法限流中："  + "metrics[ availablePermissions=" + metrics.getAvailablePermissions() +
                ", numberOfWaitingThreads=" + metrics.getNumberOfWaitingThreads() +
                " ]"
        );
        User user = new User("限流控制器降级测试", 0);
        log.info(JSONObject.toJSONString(user));
        return user;
    }
 
}

使用postman在一个限制周期内连续调用4次，控制台结果如下：

前三次能正常输出，第4次时限流器进行了限流处理，抛出异常。并且由于配置了timeOutDuration为5000ms，因此第4次接口调用时等待了5000ms才返回结果信息。

{
    "timestamp": "2022-06-01T03:44:41.563+0000",
    "status": 500,
    "error": "Internal Server Error",
    "message": "RateLimiter 'ratelimiterA' does not permit further calls",
    "trace": "io.github.resilience4j.ratelimiter.RequestNotPermitted: RateLimiter 'ratelimiterA' does not permit further calls\n\tat....",
    "path": "/ratelimiterTestNoFallbackMethod"
}

调用服务降级的限流方法ratelimiterTestFallbackMethod
使用postman在一个限制周期内连续调用4次，控制台结果如下：

{"age":27,"name":"liuxiao"}
{"age":27,"name":"liuxiao"}
{"age":27,"name":"liuxiao"}
限流控制器已经打开，拒绝访问被保护方法~
方法限流中：metrics[ availablePermissions=0, numberOfWaitingThreads=0 ]
{"age":0,"name":"限流控制器降级测试"}

八、断路器（CircuitBreaker）

断路器（CircuitBreaker）相对于前面几个熔断机制更复杂，CircuitBreaker通常存在三种状态（CLOSE、OPEN、HALF_OPEN），并通过一个时间或数量窗口来记录当前的请求成功率或慢速率，从而根据这些指标来作出正确的容错响应。

1. 当CircuitBreaker为CLOSE状态时客户端发起的请求将正常进入服务端系统，CircuitBreaker会计算出当前请求前的一个窗口里所有请求的异常率（失败率或慢速率），若异常率低于预期配置值，则系统将继续正常处理接下来的请求。
2. 当异常率不低于预期配置值时，此时服务端会进入OPEN状态，此时服务端将会暂时性的拒绝所有请求。在一段冷却时间（自定义配置）之后，服务端将自动进入HALF_OPEN状态。
3. 在半开状态(HALF_OPEN)服务端将尝试接受一定数量的请求（自定义配置），若这一定数量的请求的异常率低于预期，则此时服务端将再次恢复CLOSE状态，正常处理请求。而如果异常率还是高于预期则会继续退回到OPEN状态。

Resilience4j记录请求状态的数据结构和Hystrix不同，Hystrix是使用滑动窗口来进行存储的，而Resilience4j采用的是Ring Bit Buffer(环形缓冲区)。Ring Bit Buffer在内部使用BitSet这样的数据结构来进行存储，BitSet的结构如下图所示：

每一次请求的成功或失败状态只占用一个bit位，与boolean数组相比更节省内存。BitSet使用long[]数组来存储这些数据，意味着16个值(64bit)的数组可以存储1024个调用状态。

计算失败率需要填满环形缓冲区。例如，如果环形缓冲区的大小为10，则必须至少请求满10次，才会进行故障率的计算，如果仅仅请求了9次，即使9个请求都失败，熔断器也不会打开。但是CLOSE状态下的缓冲区大小设置为10并不意味着只会进入10个 请求，在熔断器打开之前的所有请求都会被放入。

当故障率高于设定的阈值时，熔断器状态会从由CLOSE变为OPEN。这时所有的请求都会抛出CallNotPermittedException异常。当经过一段时间后，熔断器的状态会从OPEN变为HALF_OPEN，HALF_OPEN状态下同样会有一个Ring Bit Buffer，用来计算HALF_OPEN状态下的故障率，如果高于配置的阈值，会转换为OPEN，低于阈值则装换为CLOSE。与CLOSE状态下的缓冲区不同的地方在于，HALF_OPEN状态下的缓冲区大小会限制请求数，只有缓冲区大小的请求数会被放入。

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

熔断器关于线程安全的保证措施有以下几个部分：

• 熔断器的状态使用AtomicReference保存的
• 更新熔断器状态是通过无状态的函数或者原子操作进行的
• 更新事件的状态用synchronized关键字保护
  意味着同一时间只有一个线程能够修改熔断器状态或者记录事件的状态。

6.2 可配置参数

6.3 测试demo

6.3.1 添加依赖

resilience4j-spring-boot集成了circuitbeaker、retry、bulkhead、ratelimiter几个模块，就直接引入resilience4j-spring-boot依赖。

<dependency>
    <groupId>io.github.resilience4jgroupId>
    <artifactId>resilience4j-spring-boot2artifactId>
    <version>1.2.0version>
dependency>

 6.3.2 断路器配置(application-yml)

# "order"为断路器的名字
#熔断器关闭时的缓冲区大小
resilience4j:
  circuitbreaker:
    backends:
      order:
        ringBufferSizeInClosedState: 5 # 熔断器关闭时的缓冲区大小
        ringBufferSizeInHalfOpenState: 3 # 熔断器半开时的缓冲区大小
        waitDurationInOpenState: 5000 # 熔断器从打开到半开需要的时间
        failure-rate-threshold: 60 # 熔断器打开的失败阈值
        eventConsumerBufferSize: 10 # 事件缓冲区大小
        registerHealthIndicator: true # 健康监测
        automaticTransitionFromOpenToHalfOpenEnabled: false # 是否自动从打开到半开，不需要触发

6.3.3 使用注解方式实现断路器

package com.lx.cloud.demo.controller;
 
import com.lx.cloud.demo.entity.User;
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
/**
 * Resilience4jController
 * @author lx
 */
@RestController
@Slf4j
public class Resilience4jController {
    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;
 
    @GetMapping("/circuitBreakerAOPTestNoFallbackMethod")
    @CircuitBreaker(name = "order")
    public User circuitBreakerAOPTestNoFallbackMethod() throws Exception{
        throw new Exception("服务异常");
    }
 
    @GetMapping("/circuitBreakerAOPTest")
    @CircuitBreaker(name = "order", fallbackMethod = "fallBack")
    public User circuitBreakerAOPTest() throws Exception{
        throw new Exception("服务异常");
    }
 
    private User fallBack(CallNotPermittedException e){
        log.info("熔断器已经打开，拒绝访问被保护方法~");
        io.github.resilience4j.circuitbreaker.CircuitBreaker order = circuitBreakerRegistry.circuitBreaker("order");
        io.github.resilience4j.circuitbreaker.CircuitBreaker.Metrics metrics = order.getMetrics();
        log.info("方法降级中：" + "state=" + order.getState() + " , metrics[ failureRate=" + metrics.getFailureRate() +
                ", bufferedCalls=" + metrics.getNumberOfBufferedCalls() +
                ", failedCalls=" + metrics.getNumberOfFailedCalls() +
                ", successCalls=" + metrics.getNumberOfSuccessfulCalls() +
                ", maxBufferCalls=" + metrics.getNumberOfBufferedCalls() +
                ", notPermittedCalls=" + metrics.getNumberOfNotPermittedCalls() +
                " ]"
        );
        return new  User("熔断测试", 0);
    }
 
}

调用没有服务降级处理的方法circuitBreakerAOPTestNoFallbackMethod()时，熔断前服务调用结果：

{
    "timestamp": "2022-06-01T02:48:25.998+0000",
    "status": 500,
    "error": "Internal Server Error",
    "message": "CircuitBreaker 'order' is OPEN and does not permit further calls",
    "trace": "io.github.resilience4j.circuitbreaker.CallNotPermittedException: CircuitBreaker 'order' is OPEN and does not permit further calls\n\tat ....后续太多，不做记录了可以自行实验",
    "path": "/circuitBreakerAOPTestNoFallbackMethod"
}

调用有服务降级处理的方法circuitBreakerAOPTest()时，熔断后服务调用结果：

控制台输出如下:

熔断器已经打开，拒绝访问被保护方法~
方法降级中：state=OPEN , metrics[ failureRate=100.0, bufferedCalls=3, failedCalls=3, successCalls=0, maxBufferCalls=3, notPermittedCalls=1 ]
熔断器已经打开，拒绝访问被保护方法~
方法降级中：state=OPEN , metrics[ failureRate=100.0, bufferedCalls=3, failedCalls=3, successCalls=0, maxBufferCalls=3, notPermittedCalls=2 ]

九、重试（Retry） 

重试机制比较简单，当服务端处理客户端请求异常时，服务端将会开启重试机制，重试期间内，服务端将每隔一段时间重试业务逻辑处理。 如果最大重试次数内成功处理业务，则停止重试，视为处理成功。如果在最大重试次数内处理业务逻辑依然异常，则此时系统将拒绝该请求。

十、circuitbreaker源码

---