专栏首页TECH flowerSpring Boot2+Resilience4j实现容错之Bulkhead

Spring Boot2+Resilience4j实现容错之Bulkhead

Resilience4j是一个轻量级、易于使用的容错库,其灵感来自Netflix Hystrix,但专为Java 8和函数式编程设计。轻量级,因为库只使用Vavr,它没有任何其他外部库依赖项。相比之下,Netflix Hystrix对Archaius有一个编译依赖关系,Archaius有更多的外部库依赖关系,如Guava和Apache Commons。 Resilience4j提供高阶函数(decorators)来增强任何功能接口、lambda表达式或方法引用,包括断路器、速率限制器、重试或舱壁。可以在任何函数接口、lambda表达式或方法引用上使用多个装饰器。优点是您可以选择所需的装饰器,而无需其他任何东西。 有了Resilience4j,你不必全力以赴,你可以选择你需要的。 https://resilience4j.readme.io/docs/getting-started

概览

Resilience4j提供了两种舱壁模式(Bulkhead),可用于限制并发执行的次数:

  • SemaphoreBulkhead(信号量舱壁,默认),基于Java并发库中的Semaphore实现。
  • FixedThreadPoolBulkhead(固定线程池舱壁),它使用一个有界队列和一个固定线程池。

本文将演示在Spring Boot2中集成Resilience4j库,以及在多并发情况下实现如上两种舱壁模式。

引入依赖

在Spring Boot2项目中引入Resilience4j相关依赖

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
    <version>1.4.0</version>
</dependency>
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-bulkhead</artifactId>
    <version>1.4.0</version>
</dependency>

由于Resilience4j的Bulkhead依赖于Spring AOP,所以我们需要引入Spring Boot AOP相关依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>

我们可能还希望了解Resilience4j在程序中的运行时状态,所以需要通过Spring Boot Actuator将其暴露出来

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>

实现SemaphoreBulkhead(信号量舱壁)

resilience4j-spring-boot2实现了对resilience4j的自动配置,因此我们仅需在项目中的yml/properties文件中编写配置即可。

SemaphoreBulkhead的配置项如下:

属性配置

默认值

含义

maxConcurrentCalls

25

舱壁允许的最大并行执行量

maxWaitDuration

0

尝试进入饱和舱壁时,应阻塞线程的最长时间。

添加配置

示例(使用yml):

resilience4j.bulkhead:
  configs:
    default:
      maxConcurrentCalls: 5
      maxWaitDuration: 20ms
  instances:
    backendA:
      baseConfig: default
    backendB:
      maxWaitDuration: 10ms
      maxConcurrentCalls: 20

如上,我们配置了SemaphoreBulkhead的默认配置为maxConcurrentCalls: 5,maxWaitDuration: 20ms。并在backendA实例上应用了默认配置,而在backendB实例上使用自定义的配置。这里的实例可以理解为一个方法/lambda表达式等等的可执行单元。

编写Bulkhead逻辑

定义一个受SemaphoreBulkhead管理的Service类:

@Service
public class BulkheadService {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    private BulkheadRegistry bulkheadRegistry;

    @Bulkhead(name = "backendA")
    public JsonNode getJsonObject() throws InterruptedException {
        io.github.resilience4j.bulkhead.Bulkhead.Metrics metrics = bulkheadRegistry.bulkhead("backendA").getMetrics();
        logger.info("now i enter the method!!!,{}<<<<<<{}", metrics.getAvailableConcurrentCalls(), metrics.getMaxAllowedConcurrentCalls());
        Thread.sleep(1000L);
        logger.info("now i exist the method!!!");
        return new ObjectMapper().createObjectNode().put("file", System.currentTimeMillis());
    }
}

如上,我们将@Bulkhead注解放到需要管理的方法上面。并且通过name属性指定该方法对应的Bulkhead实例名字(这里我们指定的实例名字为backendA,所以该方法将会利用默认的配置)。

定义接口类:

@RestController
public class BulkheadResource {
    @Autowired
    private BulkheadService bulkheadService;

    @GetMapping("/json-object")
    public ResponseEntity<JsonNode> getJsonObject() throws InterruptedException {
        return ResponseEntity.ok(bulkheadService.getJsonObject());
    }
}

编写测试:

首先添加测试相关依赖

<dependency>
    <groupId>io.rest-assured</groupId>
    <artifactId>rest-assured</artifactId>
    <version>3.0.5</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <version>4.0.2</version>
    <scope>test</scope>
</dependency>

这里我们使用rest-assured和awaitility编写多并发情况下的API测试

public class SemaphoreBulkheadTests extends Resilience4jDemoApplicationTests {
    @LocalServerPort
    private int port;
    @BeforeEach
    public void init() {
        RestAssured.baseURI = "http://localhost";
        RestAssured.port = port;
    }

    @Test
    public void 多并发访问情况下的SemaphoreBulkhead测试() {
        CopyOnWriteArrayList<Integer> statusList = new CopyOnWriteArrayList<>();
        IntStream.range(0, 8).forEach(i -> CompletableFuture.runAsync(() -> {
                statusList.add(given().get("/json-object").statusCode());
            }
        ));
        await().atMost(1, TimeUnit.MINUTES).until(() -> statusList.size() == 8);
        System.out.println(statusList);
        assertThat(statusList.stream().filter(i -> i == 200).count()).isEqualTo(5);
        assertThat(statusList.stream().filter(i -> i == 500).count()).isEqualTo(3);
    }
}

可以看到所有请求中只有前五个顺利通过了,其余三个都因为超时而导致接口报500异常。我们可能并不希望这种不友好的提示,因此Resilience4j提供了自定义的失败回退方法。当请求并发量过大时,无法正常执行的请求将进入回退方法。

首先我们定义一个回退方法

private JsonNode fallback(BulkheadFullException exception) {
        return new ObjectMapper().createObjectNode().put("errorFile", System.currentTimeMillis());
    }

注意:回退方法应该和调用方法放置在同一类中,并且必须具有相同的方法签名,并且仅带有一个额外的目标异常参数。

然后在@Bulkhead注解中指定回退方法:@Bulkhead(name = "backendA", fallbackMethod = "fallback")

最后修改API测试代码:

@Test
public void 多并发访问情况下的SemaphoreBulkhead测试使用回退方法() {
    CopyOnWriteArrayList<Integer> statusList = new CopyOnWriteArrayList<>();
    IntStream.range(0, 8).forEach(i -> CompletableFuture.runAsync(() -> {
            statusList.add(given().get("/json-object").statusCode());
        }
    ));
    await().atMost(1, TimeUnit.MINUTES).until(() -> statusList.size() == 8);
    System.out.println(statusList);
    assertThat(statusList.stream().filter(i -> i == 200).count()).isEqualTo(8);
}

运行单元测试,成功!可以看到,我们定义的回退方法,在请求过量时起作用了。

实现FixedThreadPoolBulkhead(固定线程池舱壁)

FixedThreadPoolBulkhead的配置项如下:

配置名称

默认值

含义

maxThreadPoolSize

Runtime.getRuntime().availableProcessors()

配置最大线程池大小

coreThreadPoolSize

Runtime.getRuntime().availableProcessors() - 1

配置核心线程池大小

queueCapacity

100

配置队列的容量

keepAliveDuration

20ms

当线程数大于核心时,这是多余空闲线程在终止前等待新任务的最长时间

添加配置

示例(使用yml):

resilience4j.thread-pool-bulkhead:
  configs:
    default:
      maxThreadPoolSize: 4
      coreThreadPoolSize: 2
      queueCapacity: 2
  instances:
    backendA:
      baseConfig: default
    backendB:
      maxThreadPoolSize: 1
      coreThreadPoolSize: 1
      queueCapacity: 1

如上,我们定义了一段简单的FixedThreadPoolBulkhead配置,我们指定的默认配置为:maxThreadPoolSize: 4,coreThreadPoolSize: 2,queueCapacity: 2,并且指定了两个实例,其中backendA使用了默认配置而backendB使用了自定义的配置。

编写Bulkhead逻辑

定义一个受FixedThreadPoolBulkhead管理的方法:

@Bulkhead(name = "backendA", type = Bulkhead.Type.THREADPOOL)
public CompletableFuture<JsonNode> getJsonObjectByThreadPool() throws InterruptedException {
    io.github.resilience4j.bulkhead.ThreadPoolBulkhead.Metrics metrics = threadPoolBulkheadRegistry.bulkhead("backendA").getMetrics();
    logger.info("now i enter the method!!!,{}", metrics);
    Thread.sleep(1000L);
    logger.info("now i exist the method!!!");
    return CompletableFuture.supplyAsync(() -> new ObjectMapper().createObjectNode().put("file", System.currentTimeMillis()));
}

如上定义和SemaphoreBulkhead的方法大同小异,其中@Bulkhead显示指定了type的属性为Bulkhead.Type.THREADPOOL,表明其方法受FixedThreadPoolBulkhead管理。由于@Bulkhead默认的Bulkhead是SemaphoreBulkhead,所以在未指定type的情况下为SemaphoreBulkhead。另外,FixedThreadPoolBulkhead只对CompletableFuture方法有效,所以我们必创建返回CompletableFuture类型的方法。

定义接口类方法

@GetMapping("/json-object-with-threadpool")
public ResponseEntity<JsonNode> getJsonObjectWithThreadPool() throws InterruptedException, ExecutionException {
    return ResponseEntity.ok(bulkheadService.getJsonObjectByThreadPool().get());
}

编写测试代码

@Test
public void 多并发访问情况下的ThreadPoolBulkhead测试() {
    CopyOnWriteArrayList<Integer> statusList = new CopyOnWriteArrayList<>();
    IntStream.range(0, 8).forEach(i -> CompletableFuture.runAsync(() -> {
            statusList.add(given().get("/json-object-with-threadpool").statusCode());
        }
    ));
    await().atMost(1, TimeUnit.MINUTES).until(() -> statusList.size() == 8);
    System.out.println(statusList);
    assertThat(statusList.stream().filter(i -> i == 200).count()).isEqualTo(6);
    assertThat(statusList.stream().filter(i -> i == 500).count()).isEqualTo(2);
}

测试中我们并行请求了8次,其中6次请求成功,2次失败。根据FixedThreadPoolBulkhead的默认配置,最多能容纳maxThreadPoolSize+queueCapacity次请求(根据我们上面的配置为6次)。

同样,我们可能并不希望这种不友好的提示,那么我们可以指定回退方法,在请求无法正常执行时使用回退方法。

private CompletableFuture<JsonNode> fallbackByThreadPool(BulkheadFullException exception) {
    return CompletableFuture.supplyAsync(() -> new ObjectMapper().createObjectNode().put("errorFile", System.currentTimeMillis()));
}
@Bulkhead(name = "backendA", type = Bulkhead.Type.THREADPOOL, fallbackMethod = "fallbackByThreadPool")
public CompletableFuture<JsonNode> getJsonObjectByThreadPoolWithFallback() throws InterruptedException {
    io.github.resilience4j.bulkhead.ThreadPoolBulkhead.Metrics metrics = threadPoolBulkheadRegistry.bulkhead("backendA").getMetrics();
    logger.info("now i enter the method!!!,{}", metrics);
    Thread.sleep(1000L);
    logger.info("now i exist the method!!!");
    return CompletableFuture.supplyAsync(() -> new ObjectMapper().createObjectNode().put("file", System.currentTimeMillis()));
}

编写测试代码

@Test
public void 多并发访问情况下的ThreadPoolBulkhead测试使用回退方法() {
    CopyOnWriteArrayList<Integer> statusList = new CopyOnWriteArrayList<>();
    IntStream.range(0, 8).forEach(i -> CompletableFuture.runAsync(() -> {
            statusList.add(given().get("/json-object-by-threadpool-with-fallback").statusCode());
        }
    ));
    await().atMost(1, TimeUnit.MINUTES).until(() -> statusList.size() == 8);
    System.out.println(statusList);
    assertThat(statusList.stream().filter(i -> i == 200).count()).isEqualTo(8);
}

由于指定了回退方法,所有请求的响应状态都为正常了。


总结

本文首先简单介绍了Resilience4j的功能及使用场景,然后具体介绍了Resilience4j中的Bulkhead。演示了如何在Spring Boot2项目中引入Resilience4j库,使用代码示例演示了如何在Spring Boot2项目中实现Resilience4j中的两种Bulkhead(SemaphoreBulkhead和FixedThreadPoolBulkhead),并编写API测试验证我们的示例。

本文示例代码地址:https://github.com/cg837718548/resilience4j-demo

欢迎访问笔者博客:blog.dongxishaonian.tech

本文分享自微信公众号 - TECH flower(tech-flower),作者:东溪陈姓少年

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2020-06-25

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Spring Cloud Circuit Breaker

    用户1215919
  • resilience4j小试牛刀

    resilience4j是一款受hystrix启发的容错组件,提供了如下几款核心组件:

    code4it
  • 实战 Spring Cloud Gateway 之限流篇

    话说在 Spring Cloud Gateway 问世之前,Spring Cloud 的微服务世界里,网关一定非 Netflix Zuul 莫属。但是由于 Zu...

    架构师修炼
  • Spring Boot 2.4.4、Spring Cloud 2020.0.2 发布

    中国古人常说,好事成双,前两天Oracle刚刚宣布了JAVA16的发布(链接),这不,Spring Boot 2.4.4和Spring Cloud 2020.0...

    程序猿DD
  • 聊聊resilience4j的bulkhead

    resilience4j-bulkhead-0.13.0-sources.jar!/io/github/resilience4j/bulkhead/Bulkhe...

    code4it
  • 图解resilience4j容错机制

    本文将介绍resilience4j中的四种容错机制,不过鉴于容错机制原理的通用性,后文所介绍的这几种容错机制也可以脱离resilience4j而独立存在(你完全...

    东溪陈姓少年
  • SpringCloud升级之路2020.0.x版-35. 验证线程隔离正确性

    上一节我们通过单元测试验证了重试的正确性,这一节我们来验证我们线程隔离的正确性,主要包括:

    干货满满张哈希
  • Spring Cloud升级之路 - Hoxton - 4. 使用Resilience4j实现实例级别的隔离与熔断

    由于我们的入口注解类从@SpringCloudApplication替换成了SpringBootApplication,这样不会启用Spring-Cloud-C...

    干货满满张哈希
  • SpringCloud升级之路2020.0.x版-33. 实现重试、断路器以及线程隔离源码

    在前面两节,我们梳理了实现 Feign 断路器以及线程隔离的思路,并说明了如何优化目前的负载均衡算法。但是如何更新负载均衡的数据缓存,以及实现重试、断路器以及线...

    干货满满张哈希
  • 2018年终总结

    整体而言,今年技术层面稍微有点拓宽,跨入了外表看上去高大上的流式计算领域,打开了另外一扇窗;而基于java的分布式/微服务领域,今年变化比较大,spring c...

    code4it
  • 新项目不要再用 Spring Cloud Hystrix 了

    Spring Cloud 之前使用的断路器是 Netfilx 开源的 Hystrix 。被很多开发人员作为默认的断路器来使用。2018 年 11 月,当 Net...

    码农小胖哥
  • Spring Boot2 系列教程(三十四)Spring Security 添加验证码

    登录添加验证码是一个非常常见的需求,网上也有非常成熟的解决方案,其实,要是自己自定义登录实现这个并不难,但是如果需要在 Spring Security 框架中实...

    江南一点雨
  • Spring Boot2 系列教程(十三)Spring Boot 中的全局异常处理

    在 Spring Boot 项目中 ,异常统一处理,可以使用 Spring 中 @ControllerAdvice 来统一处理,也可以自己来定义异常处理方案。S...

    江南一点雨
  • Spring Boot2 系列教程(四十二)使用 Jib 部署 Spring Boot

    之前松哥和大家分享过一篇将 Spring Boot 项目部署到远程 Docker 上的文章:

    江南一点雨
  • Hystrix 停止开发。。。Spring Cloud 何去何从?

    大家如果有对 Hystrix 不清楚的,请看下这篇文章:分布式服务防雪崩熔断器,Hystrix理论+实战。

    Java技术栈
  • SpringCloud升级之路2020.0.x版-6.微服务特性相关的依赖说明

    1.启用 Spring Cloud 的 Bootstrap Context:在 Spring Cloud 2020.0.x 版本开始,Bootstrap Con...

    干货满满张哈希
  • Spring Boot2 系列教程(十一)Spring Boot 中的静态资源配置

    当我们使用 SpringMVC 框架时,静态资源会被拦截,需要添加额外配置,之前老有小伙伴在微信上问松哥 Spring Boot 中的静态资源加载问题:“松哥,...

    江南一点雨
  • 《面试补习》-熔断降级我学会了!

    高可用三剑客 限流,熔断和削峰 终于来到第二篇, 熔断降级专题了,想回顾限流相关内容的童鞋,可以查看一下,下面文章,欢迎点赞,收藏,关注三连,感谢!

    九灵
  • Spring Boot2 系列教程(十四)CORS 解决跨域问题

    很多人对跨域有一种误解,以为这是前端的事,和后端没关系,其实不是这样的,说到跨域,就不得不说说浏览器的同源策略。

    江南一点雨

扫码关注云+社区

领取腾讯云代金券