SpringCloud之Resilience4J用法精讲

向右看齐 2021-11-04 14:46 506阅读 0赞

在微服务中，经常会出现一些故障，而一些故障会直接或者间接的拖垮其它的服务，造成服务器雪崩，系统就会死掉。

什么是服务雪崩？我们可以通过下面一张图来看：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70]  
假如现在有很多的用户同时请求订单微服务去执行下单的操作，那么会调用我们的支付微服务，如果支付微服务现在挂掉了，而订单调用一直没有响应，由于很多的用户执行相同的操作，属于高并发，那么服务器上积累的订单越来越多，那么原来没有问题的订单微服务，也会被拖垮，这就是服务雪崩。

我们需要做的就是，当某一个微服务发生蔓延当时候，不能发生故障蔓延，整个系统还能以其它某种方式正常运行，这个就是我们需要解决的。

# 断路器 #

我们耳熟能详的就是Netflix Hystrix,这个断路器是SpringCloud中最早支持的一种容错方案，现在这个断路器已经处于维护状态，已经不再更新了，你仍然可以使用这个断路器，但是呢，我不建议你去使用，因为这个已经不再更新，所以Spring官方已经出现了Netflix Hystrix的替换方案。 如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 1]  
在 Spring Cloud Greenwich 版中，对于 Hystrix 以及 Hystrix Dashboard 官方都给出了替代方案。我们整个教程虽然基于最新的 Spring Cloud Greenwich 版，但是考虑到现实情况，本文中我还是先向大家大致介绍一下 Hystrix 的功能，后面我们会详细介绍 Resilience4j 的用法。

# 服务熔断 #

什么是服务熔断呢？服务熔断就是当A服务去调用B服务，如果A服务迟迟没有收到B服务的响应，那么就终断当前的请求，而不是一直等待下去，一直等待下去的结果就是拖垮其它的服务。当系统发生熔断的时候，我们还要去监控B服务，当B服务恢复正常使用时，A服务就发起重新调用的请求。

# 服务降级 #

当我们的服务发生熔断的时候，那么就需要降级了，那么什么是降级？降级指的是A服务调用B服务，没有调用成功，发生熔断，那么A服务就不要死板的一直请求B服务，而是去服务上哪一个缓存先顶着，避免给我们的用户，响应一些错误的页面，这个就是服务降级。

# 请求缓存 #

请求缓存是指对接口进行缓存，这样可以大大降低服务提供者的压力，当然我们要选择缓存使用的场景，是那种更新频率低，但是访问又比较频繁的数据。

我们这里说的缓存啊，指的是Hystrix的缓存，但是在实际的开发中，我们可能会配合其它的缓存来实现更好的效果，如redis。

# 请求合并 #

我们知道SpringCloud中的微服务之间的调用都是通过HTTP来实现的，但是我们通过HTTP协议调用微服务时候，如果是高并发数量小的话，那么效率很低，那么我们可以通过合并请求来实现，也就是将客户端多个请求合并成一个请求，也就是只发送一个HTTP请求，服务器拿到请求结果后，再将请求结果分发给不同的请求，这样就可以提供传输效率。

# Resilience4J #

Resilience4J是我们Spring Cloud G版本 推荐的容错方案，它是一个轻量级的容错库。它呢借鉴了Hystrix而设计，并且采用JDK8 这个函数式编程，也就是我们的lambda表达式，为什么说它是轻量级的呢？因为它的库只使用 Vavr （以前称为 Javaslang ），它没有任何其他外部库依赖项。相比之下， Netflix Hystrix 对Archaius 具有编译依赖性，这导致了更多的外部库依赖，例如 Guava 和 Apache Commons 。而如果使用Resilience4j，你无需引用全部依赖，可以根据自己需要的功能引用相关的模块即可。

Resilience4J 提供了一系列增强微服务的可用性功能：

1.  断路器
2.  限流
3.  基于信号量的隔离
4.  缓存
5.  限时
6.  请求重启

那么我们接下来就讲解Resilience4J的几种功能的使用方法，由于是基本用法所以我们不用创建SpringBoot工程，我们只需要创建一个叫Resilience4j的普通maven工程并加入 junit 单元测试依赖，这样准备工作就完成了。

junit单元测试

<dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
      </dependency>

# 断路器初始化 #

我们使用的是Resilience4J 提供的断路器功能，那么就要加入依赖

<dependency>
             <groupId>io.github.resilience4j</groupId>
             <artifactId>resilience4j-circuitbreaker</artifactId>
             <version>0.13.2</version>
      </dependency>

这个依赖它提供的是一个基于ConcurrentHashMap 的 CircuitBreakerRegistry ，CircuitBreakerRegistry 是线程安全的，并且是原子操作。开发者可以使用 CircuitBreakerRegistry 来创建和检索 CircuitBreaker 的实例 ，开发者可以直接使用默认的全局CircuitBreakerConfig 为所有 CircuitBreaker 实例创建 CircuitBreakerRegistry ，如下所示：

CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults();

我们使用自定义的CircuitBreakerConfig，可以配置如下参数：

*  故障率阈值百分比，超过这个阈值，断路器就会打开
 *  断路器保持打开的时间，在到达设置的时间之后，断路器会进入到 half open 状态
 *  当断路器处于 half open 状态时，环形缓冲区的大小
 *  当断路器关闭时，环形缓冲区的大小
 *  自定义断路器中的事件操作
 *  自定义 Predicate 以便计算异常是否被记录为失败事件
 *  具体的定义如下：
    
        CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
            .failureRateThreshold(50)
            .waitDurationInOpenState(Duration.ofMillis(1000))
            .ringBufferSizeInHalfOpenState(2)
            .ringBufferSizeInClosedState(2)
            .build();
        CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.of(circuitBreakerConfig);
        CircuitBreaker circuitBreaker2 = circuitBreakerRegistry.circuitBreaker("otherName");
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("uniqueName", circuitBreakerConfig);
    
    上面这段代码解释如下：  
    我们首先定义了一个CircuitBreakerConfig 对象，在定义CircuitBreakerConfig对象时，设置故障率为50%，断路器 保持打开时间为2秒，断路器处于half open的时候，缓冲区大小为2，当对象处于关闭时，缓冲区的大小也是2，然后根据CircuitBreakerConfig对象创建  
    CircuitBreakerRegistry，然后再根据CircuitBreakerRegistry 创建两个断路器CircuitBreaker。
    
    如果不想使用CircuitBreakerRegistry来管理断路器 那么可以直接创建CircuitBreaker对象
    
        CircuitBreaker defaultCircuitBreaker = CircuitBreaker.ofDefaults("testName");
        CircuitBreaker customCircuitBreaker = CircuitBreaker.of("testName", circuitBreakerConfig);
    
    # 断路器的使用案例 #
    
    断路器使用了装饰者模式，开发者可以使用 CircuitBreaker.decorateCheckedSupplier(), CircuitBreaker.decorateCheckedRunnable() 或者 CircuitBreaker.decorateCheckedFunction() 来装饰 Supplier / Runnable / Function 或者 CheckedRunnable / CheckedFunction，然后使用 Try.of(…) 或者 Try.run(…) 来进行调用操作，也可以使用 map、flatMap、filter、recover 或者 andThen 进行链式调用，但是调用这些方法断路器必须处于 CLOSED 或者 HALF\_OPEN 状态。例如下面一个例子，创建一个断路器出来，首先装饰了一个函数，这个函数返回一段字符串，然后使用 Try.of 去执行，执行完后再进入到 map 中去执行。如果第一个函数正常执行第二个函数才会执行，如果第一个函数执行失败，那么 map 函数将不会执行：
    
        CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("testName");
        CheckedFunction0<String> decoratedSupplier = CircuitBreaker
                .decorateCheckedSupplier(circuitBreaker, () -> "This can be any method which returns: 'Hello");
        Try<String> result = Try.of(decoratedSupplier)
                .map(value -> value + " world'");
        System.out.println(result.isSuccess());
        System.out.println(result.get());
    
    你可以将不同的断路器连接起来：
    
        CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("testName");
        CircuitBreaker anotherCircuitBreaker = CircuitBreaker.ofDefaults("anotherTestName");
        CheckedFunction0<String> decoratedSupplier = CircuitBreaker
                .decorateCheckedSupplier(circuitBreaker, () -> "Hello");
        CheckedFunction1<String, String> decoratedFunction = CircuitBreaker
                .decorateCheckedFunction(anotherCircuitBreaker, (input) -> input + " world");
        Try<String> result = Try.of(decoratedSupplier)
                .mapTry(decoratedFunction::apply);
        System.out.println(result.isSuccess());
        System.out.println(result.get());
    
    # 断路器的打开 #
    
    这里创建了两个 CircuitBreaker ，装饰了两个函数，第二次使用了 mapTry 方法来连接。前面给大家演示的几种情况，都是执行成功的，即断路器一直处于关闭的状态，接下来给大家再来演示一个断路器打开的例子，如下：
    
        CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
                .ringBufferSizeInClosedState(2)
                .waitDurationInOpenState(Duration.ofMillis(1000))
                .build();
        CircuitBreaker circuitBreaker = CircuitBreaker.of("testName", circuitBreakerConfig);
        circuitBreaker.onError(0, new RuntimeException());
        System.out.println(circuitBreaker.getState());
        circuitBreaker.onError(0, new RuntimeException());
        System.out.println(circuitBreaker.getState());
        Try<String> result = Try.of(CircuitBreaker.decorateCheckedSupplier(circuitBreaker, () -> "Hello"))
                .map(value -> value + " world");
        System.out.println(result.isSuccess());
        System.out.println(result.get());
    
    这里手动模拟错误，首先设置了断路器关闭状态下的环形缓冲区大小为 2 ，即当有两条数据时就可以去统计故障率了，这里没有设置故障率，默认的故障率是 50% ，当第一次调用 onError 方法后，打印断路器当前状态，发现断路器还是处于关闭状态，并未打开，接下来再次调用 onError 方法，然后再去查看断路器状态，此时发现断路器已经打开了，因为满足了 50% 的故障率了。
    
    # 断路器重置 #
    
    断路器的重置就是清空数据
    
        circuitBreaker.reset();
    
    # 服务器请求降级 #
    
    服务器降级操作如下：
    
        CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("testName");
        CheckedFunction0<String> checkedSupplier = CircuitBreaker.decorateCheckedSupplier(circuitBreaker, () -> { 
            throw new RuntimeException("BAM!");
        });
        Try<String> result = Try.of(checkedSupplier)
                .recover(throwable -> "Hello Recovery");
        System.out.println(result.isSuccess());
        System.out.println(result.get());
    
    如果需要使用服务降级，可以使用 Try.recover() 链接，当 Try.of() 返回 Failure 时服务降级会被触发。
    
    # 状态监听 #
    
    状态监听可以获取到熔断器当前的运行数据，例如：
    
        CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();
        // 获取故障率
        float failureRate = metrics.getFailureRate();
        // 获取调用失败次数
        int failedCalls = metrics.getNumberOfFailedCalls();
    
    # 限流 #
    
    RateLimiter 和我们前面提到的断路器实际上非常类似，它也有一个基于内存的 RateLimiterRegistry 和 RateLimiterConfig 可以配置，我们可以配置如下一些参数：
 *  限流之后的冷却时间
 *  阈值刷新时间
 *  阈值刷新频次
 *  使用限流我们要引入下面的依赖：
    
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-ratelimiter</artifactId>
            <version>0.13.2</version>
        </dependency>
    
    **基本用法**
    
    例如，想限制某个请求的频率为 2QPS（每秒处理两个请求），为什么给一个这样的频率呢？主要是为了大家一会儿测试方便，代码如下：
    
        RateLimiterConfig config = RateLimiterConfig.custom()
                .limitRefreshPeriod(Duration.ofMillis(1000))
                .limitForPeriod(2)
                .timeoutDuration(Duration.ofMillis(1000))
                .build();
        RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(config);
        RateLimiter rateLimiterWithDefaultConfig = rateLimiterRegistry.rateLimiter("backend");
        RateLimiter rateLimiterWithCustomConfig = rateLimiterRegistry.rateLimiter("backend#2", config);
        RateLimiter rateLimiter = RateLimiter.of("NASDAQ :-)", config);
    
    和前面的一样，我们也可以使用 RateLimiterRegistry 来统一管理 RateLimiter ，也可以通过 RateLimiter.of 方法来直接创建一个 RateLimiter。创建好了，就可以直接使用了，代码如下：
    
        CheckedRunnable restrictedCall = RateLimiter
                .decorateCheckedRunnable(rateLimiter,()->{ 
                    System.out.println(new Date());
                });
        Try.run(restrictedCall)
                .andThenTry(restrictedCall)
                .andThenTry(restrictedCall)
                .andThenTry(restrictedCall)
                .onFailure(throwable -> System.out.println(throwable.getMessage()));
    
    执行结果如下：
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 2]  
    可以观察上面，我们发现可以知道限流一次执行了两个方法，另外两个方法在1s过后执行的。并且限流参数是可以随便修改的，修改后，本次的限流周期内不会生效，下次限流才会生效执行。
    
    修改限流如下：
    
        rateLimiter.changeLimitForPeriod(100);
        rateLimiter.changeTimeoutDuration(Duration.ofMillis(100));
    
    # 事件监听 #
    
    在限流中，我们可以获取所有允许和拒绝执行的事件信息，获取方式如下：
    
        rateLimiter.getEventPublisher()
                .onSuccess(event -> { 
                    System.out.println(new Date()+">>>"+event.getEventType()+">>>"+event.getCreationTime());
                })
                .onFailure(event -> { 
                    System.out.println(new Date()+">>>"+event.getEventType()+">>>"+event.getCreationTime());
                });
    
    # 请求隔离 #
    
    这里的请求隔离,主要是基于信号量的请求隔离，不包含基于线程的请求隔离，具体用法和前面两个类似，不过在使用之前，需要先添加请求隔离相关的依赖，如下：
    
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-bulkhead</artifactId>
            <version>0.13.2</version>
        </dependency>
    
    定义最大并行数和饱和状态Bulkhead时 线程的最大阻塞时间 如下：
    
        BulkheadConfig config = BulkheadConfig.custom()
                        .maxConcurrentCalls(150)
                        .maxWaitTime(100)
                        .build();
                BulkheadRegistry registry = BulkheadRegistry.of(config);
                Bulkhead bulkhead1 = registry.bulkhead("foo");
                BulkheadConfig custom = BulkheadConfig.custom()
                        .maxWaitTime(0).build();
        
        
                Bulkhead bulkhead2 = registry.bulkhead("bar", custom);
                System.out.println(bulkhead1 + ">>>>>" + bulkhead2);
            }
    
    输出结果为：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 3]
    
    如果不想通过BulkheadRegistry来管理Bulkhead的实例，那么我们可以直接创建Bulkhead如下：
    
        Bulkhead bulkhead1 = Bulkhead.ofDefaults("foo");
                Bulkhead bulkhead2 = Bulkhead.of("bar",
                        BulkheadConfig.custom().maxConcurrentCalls(50).build());
                System.out.println(bulkhead1 + ">>>>>" + bulkhead2);
    
    运行效果如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 4]  
    创建好来后，使用的方法和断路器是一样的：
    
        BulkheadConfig config = BulkheadConfig.custom()
                        .maxConcurrentCalls(1)
                        .maxWaitTime(100)
                        .build();
                Bulkhead bulkhead = Bulkhead.of("testName", config);
                CheckedFunction0<String> decoratedSupplier = Bulkhead.decorateCheckedSupplier(
                        bulkhead,()-> "this is bulkhead: love "
                );
        
                Try<String> result = Try.of(decoratedSupplier).map(
                        value -> value + "小蕾"
                );
        
                System.out.println(result.isSuccess());
                System.out.println(result.get());
            }
    
    执行结果如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 5]
    
    # 请求重试 #
    
    当我们的服务失败的时候，那么就需要请求重试，可以说请求重试是一个非常常用的功能，要使用请求 重试的话，那么要引入下面这个依赖：
    
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-retry</artifactId>
            <version>0.13.2</version>
        </dependency>
    
    那么引入依赖后，我们创建一个重试的测试用例：
    
        RetryConfig config = RetryConfig.custom()
           .maxAttempts(3) //重试次数为3次
           .waitDuration(Duration.ofMillis(500)) //每次重试间隔500毫秒
           .build();
          Retry retry = Retry.of("id", config);
    
    创建好 Retry的实例后，我们就可以使用了，使用的步骤和断路器是一样的如下：
    
        CheckedFunction0<String> retryAllSupplier = Retry.decorateCheckedSupplier(
                        retry, () -> { 
                            System.out.println("date:" + new Date() + ":" + Math.random());
                            return "love 小蕾";
                        });
        
                Try<String> result = Try.of(retryAllSupplier).recover((throwable -> "Hello world from recovery function"));
                System.out.println(result.isSuccess());
                System.out.println(result.get());
    
    运行的结果如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 6]
    
    如果抛出了异常那么就会触发重试机制。
    
    # 缓存 #
    
    Resilience4J 提供了 JCache 缓存，但是我们实际开发用的是Redis缓存，这里就不多讲了，感兴趣的朋友，可以自己去学习。
    
    # 限时 #
    
    Resilience4j 中的限时器是要结合 Future 一起来使用，开发者需要提前配置过期时间，在过期时间内要是没有获取到value，那么 Future 将会被取消，使用步骤如下：  
    先引入依赖
    
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-timelimiter</artifactId>
            <version>0.13.2</version>
        </dependency>
    
    使用代码如下：
    
        TimeLimiterConfig config = TimeLimiterConfig.custom()
                        .timeoutDuration(Duration.ofSeconds(60))
                        .cancelRunningFuture(true)
                        .build();
                TimeLimiter timeLimiter = TimeLimiter.of(config);
                ExecutorService executorService = Executors.newSingleThreadExecutor();
                Supplier<Future<Integer>> futureSupplier = () -> executorService.submit(userService::doSomething);
                Callable restrictedCall = TimeLimiter
                        .decorateFutureSupplier(timeLimiter, futureSupplier);
                Try.of(restrictedCall.call)
                        .onFailure(throwable -> System.out.println(throwable.getMessage()));
    
    这里首先创建了一个 TimeLimiter，然后将任务放到线程池中，获取到一个 Supplier 对象，然后使用限时器包装该对象，当调用超时， onFailure 方法就会被触发。
    
    也可以将限时器和断路器结合使用，当调用超时次数过多，直接熔断，如下：
    
        Callable restrictedCall = TimeLimiter
            .decorateFutureSupplier(timeLimiter, futureSupplier);
        Callable chainedCallable = CircuitBreaker.decorateCallable(circuitBreaker, restrictedCall);
        Try.of(chainedCallable::call)
            .onFailure(throwable -> LOG.info("We might have timed out or the circuit breaker has opened."));
    
    # 总结 #
    
    本文首先向大家介绍了传统的容错方案 Hystrix 的一些大致功能，这个读者作为了解即可；然后向读者介绍了 Resilience4j 的一些基本功能，这些基本功能涵盖了请求熔断、限流、限时、缓存、隔离以及重试，这里我们只是介绍了 Resilience4j 的一些基本用法。上文中所有的案例都是在一个普通的 JavaSE 项目中写的，这里并未涉及到微服务，下篇文章我将和大家分享，这六个功能如何在微服务中使用，进而实现微服务系统的高可用。
    
    # 项目地址 #
    
    [github][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70]: /images/20211104/97c3b12fbdc24095a3c0af0f6be536a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 1]: /images/20211104/8b3fdda2e4bb49e2bfa08aa81ff81e3b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 2]: /images/20211104/83ba22718bbc4c788003f33f0b70c863.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 3]: /images/20211104/532fa0f8eb344185a0110230e4491a5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 4]: /images/20211104/fc117cd5038c4b46ac1ddf7ecac29305.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 5]: /images/20211104/8f125c10764e49f5818c247b5f47cb7b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3ZTg2MzE0_size_16_color_FFFFFF_t_70 6]: /images/20211104/de4c26d077604174bbf8a4fe796c631d.png
[github]: https://github.com/zhoubiao188/spring_cloud_study/tree/master/Resilience4j