Resilience4J 熔断器

简介

Resilience4j 主要有以下功能：

• CircuitBreaker(熔断器)
• RateLimiter(限流)
• Retry(请求重试)
• 限时
• 缓存
• 信号量的隔离

常用熔断器比较

简单使用

熔断器

• pom.xml 添加依赖

        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-circuitbreaker</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

• 创建测试用例测试短路器的功能

 @Test
    public void testCircuitBreaker() {
        // 创建断路器配置
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                //故障率阈值百分比(百分之50)，超过这个阈值，断路器就会打开
                .failureRateThreshold(50)
                //断路器保持打开的时间，在到达设置的时间之后，断路器会进入到 half open 状态
                .waitDurationInOpenState(Duration.ofMillis(1000))
                //当断路器处于half open 状态时，环形缓冲区的大小
                 .ringBufferSizeInHalfOpenState(2)//半打开次数
                .ringBufferSizeInClosedState(2)//至少要调用两次
                .build();
        //按以上配置创建断路器实例
        CircuitBreakerRegistry r1 = CircuitBreakerRegistry.of(config);
        CircuitBreaker cb1 = r1.circuitBreaker("yuan11");//创建默认断路器
        CircuitBreaker cb2 = r1.circuitBreaker("yuan22", config);// 创建上述配置方案的断路器
        // 测试调用
        CheckedFunction0<String> supplier = CircuitBreaker.decorateCheckedSupplier(cb1, () -> "hello resilience4j");
        Try<String> result = Try.of(supplier)
                .map(v -> v + " hello world");
        System.out.println(result.isSuccess());
        System.out.println(result.get());
    }

• 运行测试用例 结果如下：
  
  可以看到正常情况调用了短路请求正常 未触发。
• 手动触发短路器

@Test
    public void testCircuitBreaker2() {
        // 创建断路器配置
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                //故障率阈值百分比(百分之50)，超过这个阈值，断路器就会打开
                .failureRateThreshold(50)
                //断路器保持打开的时间，在到达设置的时间之后，断路器会进入到 half open 状态
                .waitDurationInOpenState(Duration.ofMillis(1000))
                //当断路器处于half open 状态时，环形缓冲区的大小
                 .ringBufferSizeInHalfOpenState(2)//半打开次数
                .ringBufferSizeInClosedState(2)//至少要调用两次
                .ignoreExceptions(NullPointerException.class)// 忽略空指针异常
                .build();
        //按以上配置创建断路器实例
        CircuitBreakerRegistry r1 = CircuitBreakerRegistry.of(config);
        CircuitBreaker cb1 = r1.circuitBreaker("yuan11");
        //断路器开始状态
        System.out.println(cb1.getState());
        //第一次调用 1s
        cb1.onError(1, TimeUnit.SECONDS, new NullPointerException());
        //获取断路器的第一次状态
        System.out.println(cb1.getState());
        // 第二次调用1m
        cb1.onError(1,TimeUnit.SECONDS, new NullPointerException());
        //获取断路器的第二次状态
        System.out.println(cb1.getState());

    }

• 运行结果如下：
  

• 运行结果如下：
  

• 重置断路器

        cb1.reset();
        System.out.println(cb1.getState());

RateLimiter 限流

• 执行测试方法

    @Test
    public void testRateLimiter (){
        RateLimiterConfig build = RateLimiterConfig.custom()
                // 阈值刷新的时间 1 秒
                .limitRefreshPeriod(Duration.ofMillis(1000))
                // 限制频次
                .limitForPeriod(2)
                // 限流之后的冷却时间 1秒
                .timeoutDuration(Duration.ofMillis(1000))
                .build();
        RateLimiter limiter = RateLimiter.of("test", build);

        CheckedRunnable runnable = RateLimiter.decorateCheckedRunnable(limiter, () -> {
            System.out.println(new Date());
        });

        // 执行4次
        Try.run(runnable)
                .andThenTry(runnable)
                .andThenTry(runnable)
                .andThenTry(runnable)
                .onFailure(t -> System.out.println(t.getMessage()));
    }

• 运行结果 可以看到2次之后，间隔了一秒重新执行
  

请求重试 Retry

• 修改pom.xml 引入依赖

        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-retry</artifactId>
        </dependency>

• 执行测试用例：

@Test
    public void testRetry(){
        RetryConfig config = RetryConfig.custom()
                // 重试次数 【小于等于下面的count次数就抛出异常】
                .maxAttempts(4)
                // 重试间隔
                .waitDuration(Duration.ofMillis(500))
                // 重试异常
                .retryExceptions(RuntimeException.class)
                .build();

        Retry retry = Retry.of("testRetry", config);
        Retry.decorateRunnable(retry, new Runnable() {
            int count = 0;
            // 重试功能开启后 执行run方法 若抛出异常 会自动触发重试功能
            @Override
            public void run() {
                if (count++ < 4){
                    System.out.println(count);
                    throw new RuntimeException();
                }
            }
        }).run();
    }

• 运行结果如下：抛出异常后执行了4次重试
  

限时 Timelimiter

• 执行测试用例

    @Test
    public void testTimelimiter(){
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        TimeLimiterConfig config = TimeLimiterConfig.custom()
                .timeoutDuration(Duration.ofMillis(600))// 超时时间
                .cancelRunningFuture(true)//是否取消线程执行
                .build();
        TimeLimiter timeLimiter = TimeLimiter.of(config);

        Supplier<Future<String>> futureSupplier = () -> {
            return executorService.submit(new testRunable());
        };
        Callable<String> restrictedCall = TimeLimiter.decorateFutureSupplier(timeLimiter,futureSupplier);
        Try.of(restrictedCall::call)
                .onFailure(throwable -> System.out.println("We might have timed out or the circuit breaker has opened."));
    }
    class testRunable implements Callable {

        @Override
        public Object call() {
            for (int i = 0; i < 4; i++) {
                System.out.println(i);
                throw  new RuntimeException();
            }
            return null;
        }
    }

• 运行结果如下: 4次循环第一次抛出异常之后就不再执行取消了线程执行。
  

请求隔离bulkhead

• pom.xml 添加依赖

        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-bulkhead</artifactId>
        </dependency>

• 执行测试用例

    @Test
    public void testBulkHead(){
        BulkheadConfig config = BulkheadConfig.custom()
                .maxConcurrentCalls(2)// 最大并发
                .maxWaitDuration(Duration.ofSeconds(1))//获取信号量的最长排队等待时间
                //.fairCallHandlingStrategyEnabled(true)// 公平锁策略，等待线程竞争
                .writableStackTraceEnabled(true)//BulkheadFullException 发生时减少堆栈跟踪中的信息量
                .build();
        BulkheadRegistry registry = BulkheadRegistry.of(config);
        Bulkhead bulkhead = registry.bulkhead("foo");
        Runnable decoratedSupplier = Bulkhead.decorateRunnable(bulkhead,()->{
            System.out.println("test");
            bulkhead.acquirePermission();
        });

        for (int i=0; i<4; i++) {
            CompletableFuture.runAsync(decoratedSupplier)
                    .thenAccept(flights -> System.out.println("Received results"));
        }
    }

• 执行结果如下：最大并发是2，则获取了两次权限之后，不并发了执行完毕了。？？