Spring Cloud Gateway自定义过滤器实战(观测断路器状态变化)
阅读原文时间:2021年11月23日阅读:1

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;

本篇概览

  • 本文是《Spring Cloud Gateway实战》系列的第七篇,前面的文章咱们学习了各种内置过滤器,还在《Spring Cloud Gateway的断路器(CircuitBreaker)功能》一文深入研究了断路器类型的过滤器(理论&实战&源码分析皆有),相信聪明的您一定会有此疑问:内置的再多也无法覆盖全部场景,定制才是终极武器

  • 所以今天咱们就来开发一个自己专属的过滤器,至于此过滤器的具体功能,其实前文已埋下伏笔,如下图:

  • 简单来说,就是在一个有断路器的Spring Cloud Gateway应用中做个自定义过滤器,在处理每个请求时把断路器的状态打印出来,这样咱们就能明明白白清清楚楚知道断路器的状态啥时候改变,变成了啥样,也算补全了《Spring Cloud Gateway的断路器(CircuitBreaker)功能》的知识点

  • 过滤器分为全局和局部两种,这里咱们选用局部的,原因很简单:咱们的过滤器是为了观察断路器,所以不需要全局生效,只要在使用断路器的路由中生效就够了;

套路提前知晓

  • 咱们先看看自定义局部过滤器的的基本套路:
  1. 新建一个类(我这里名为StatePrinterGatewayFilter.java),实现GatewayFilter和Ordered接口,重点是filter方法,该过滤器的主要功能就在这里面实现
  2. 新建一个类(我这里名为StatePrinterGatewayFilterFactory.java),实现AbstractGatewayFilterFactory方法,其apply方法的返回值就是上一步新建的StatePrinterGatewayFilter的实例,该方法的入参是在路由配置中过滤器节点下面的配置,这样就可以根据配置做一些特殊的处理,然后再创建实例作为返回值
  3. StatePrinterGatewayFilterFactory类实现String name()方法,该方法的返回值就是路由配置文件中过滤器的name
  4. String name()也可以不实现,这是因为定义该方法的接口中有默认实现了,如下图,这样您在路由配置文件中过滤器的name只能是StatePrinter:

  1. 在配置文件中,添加您自定义的过滤器,该操作和之前的添加内置过滤器一模一样
  • 以上就是自定义过滤器的基本套路了,可见还是非常简单的,接下来的实战也是按照这个套路来的

  • 在编写自定义过滤器代码之前,还有个拦路虎等着我们,也就是咱们过滤器的基本功能:如何取得断路器的状态

如何取得断路器的状态

  • 前文的代码分析中,咱们了解到断路器的核心功能集中在SpringCloudCircuitBreakerFilterFactory.apply方法中(没错,就是刚才提到的apply方法),打开这个类,如下图,从绿框可见断路器功能来自名为cb的对象,而这个对象是在红框处由reactiveCircuitBreakerFactory创建的:

  • 展开上图红框右侧的reactiveCircuitBreakerFactory.create方法继续看,最终跟踪到了ReactiveResilience4JCircuitBreakerFactory类,发现了一个极其重要的变量,就是下图红框中的circuitBreakerRegistry,它的内部有个ConcurrentHashMap(InMemoryRegistryStore的entryMap),这里面存放了所有断路器实例:

  • 此时您应该想到了,拿到断路器的关键就是拿到上图红框中的circuitBreakerRegistry对象,不过怎么拿呢?首先它是私有类型的,其次虽然有个方法返回了该对象,但是此方法并非public的,如下图红框:

  • 这个问题当然难不倒聪明的您了,没错,用反射修改此方法的访问权限,稍后的代码中咱们就这么干

  • 还剩最后一个问题:circuitBreakerRegistry是ReactiveResilience4JCircuitBreakerFactory的成员变量,这个ReactiveResilience4JCircuitBreakerFactory从哪获取?

  • 如果您配置过断路器,对这个ReactiveResilience4JCircuitBreakerFactory就很熟悉了,设置该对像是配置断路器的基本操作,回顾一下前文的代码:

    @Configuration
    public class CustomizeCircuitBreakerConfig {

    @Bean
    public ReactiveResilience4JCircuitBreakerFactory defaultCustomizer() {
    CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom() //
            .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.TIME_BASED) // 滑动窗口的类型为时间窗口
            .slidingWindowSize(10) // 时间窗口的大小为60秒
            .minimumNumberOfCalls(5) // 在单位时间窗口内最少需要5次调用才能开始进行统计计算
            .failureRateThreshold(50) // 在单位时间窗口内调用失败率达到50%后会启动断路器
            .enableAutomaticTransitionFromOpenToHalfOpen() // 允许断路器自动由打开状态转换为半开状态
            .permittedNumberOfCallsInHalfOpenState(5) // 在半开状态下允许进行正常调用的次数
            .waitDurationInOpenState(Duration.ofSeconds(5)) // 断路器打开状态转换为半开状态需要等待60秒
            .recordExceptions(Throwable.class) // 所有异常都当作失败来处理
            .build();
    
    ReactiveResilience4JCircuitBreakerFactory factory = new ReactiveResilience4JCircuitBreakerFactory();
    factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
            .timeLimiterConfig(TimeLimiterConfig.custom().timeoutDuration(Duration.ofMillis(200)).build())
            .circuitBreakerConfig(circuitBreakerConfig).build());
    
    return factory;
    }

    }

  • 既然ReactiveResilience4JCircuitBreakerFactory是spring的bean,那我们在StatePrinterGatewayFilterFactory类中用Autowired注解就能随意使用了

  • 至此,理论分析已全部完成,问题都已经解决,开始编码

源码下载

名称

链接

备注

项目主页

https://github.com/zq2599/blog_demos

该项目在GitHub上的主页

git仓库地址(https)

https://github.com/zq2599/blog_demos.git

该项目源码的仓库地址,https协议

git仓库地址(ssh)

git@github.com:zq2599/blog_demos.git

该项目源码的仓库地址,ssh协议

  • 这个git项目中有多个文件夹,本篇的源码在spring-cloud-tutorials文件夹下,如下图红框所示:

  • spring-cloud-tutorials文件夹下有多个子工程,本篇的代码是circuitbreaker-gateway,如下图红框所示:

编码

  • 前文创建了子工程circuitbreaker-gateway,此工程已添加了断路器,现在咱们的过滤器代码就写在这个工程中是最合适的了

  • 接下来按照套路写代码,首先是StatePrinterGatewayFilter.java,代码中有详细注释就不再啰嗦了,要注意的是getOrder方法返回值是10,这表示过滤器的执行顺序:

    package com.bolingcavalry.circuitbreakergateway.filter;

    import io.github.resilience4j.circuitbreaker.CircuitBreaker;
    import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
    import io.vavr.collection.Seq;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.cloud.circuitbreaker.resilience4j.ReactiveResilience4JCircuitBreakerFactory;
    import org.springframework.cloud.gateway.filter.GatewayFilter;
    import org.springframework.cloud.gateway.filter.GatewayFilterChain;
    import org.springframework.core.Ordered;
    import org.springframework.http.HttpStatus;
    import org.springframework.web.server.ServerWebExchange;
    import reactor.core.publisher.Mono;
    import java.lang.reflect.Method;

    public class StatePrinterGatewayFilter implements GatewayFilter, Ordered {

    private ReactiveResilience4JCircuitBreakerFactory reactiveResilience4JCircuitBreakerFactory;
    
    // 通过构造方法取得reactiveResilience4JCircuitBreakerFactory实例
    public StatePrinterGatewayFilter(ReactiveResilience4JCircuitBreakerFactory reactiveResilience4JCircuitBreakerFactory) {
        this.reactiveResilience4JCircuitBreakerFactory = reactiveResilience4JCircuitBreakerFactory;
    }
    
    private CircuitBreaker circuitBreaker = null;
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 这里没有考虑并发的情况,如果是生产环境,请您自行添加上锁的逻辑
        if (null==circuitBreaker) {
            CircuitBreakerRegistry circuitBreakerRegistry = null;
            try {
                Method method = reactiveResilience4JCircuitBreakerFactory.getClass().getDeclaredMethod("getCircuitBreakerRegistry",(Class[]) null);
                // 用反射将getCircuitBreakerRegistry方法设置为可访问
                method.setAccessible(true);
                // 用反射执行getCircuitBreakerRegistry方法,得到circuitBreakerRegistry
                circuitBreakerRegistry = (CircuitBreakerRegistry)method.invoke(reactiveResilience4JCircuitBreakerFactory);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
        // 得到所有断路器实例
        Seq&lt;CircuitBreaker&gt; seq = circuitBreakerRegistry.getAllCircuitBreakers();
        // 用名字过滤,myCircuitBreaker来自路由配置中
        circuitBreaker = seq.filter(breaker -&gt; breaker.getName().equals("myCircuitBreaker"))
                .getOrNull();
    }
    
    // 取断路器状态,再判空一次,因为上面的操作未必能取到circuitBreaker
    String state = (null==circuitBreaker) ? "unknown" : circuitBreaker.getState().name();
    
    System.out.println("state : " + state);
    
    // 继续执行后面的逻辑
    return chain.filter(exchange);
    } @Override public int getOrder() { return 10; }

    }

  • 接下来是StatePrinterGatewayFilterFactory.java,这里用不上什么配置,所以apply方法的入参也就没用上,需要注意的是通过Autowired注解拿到了reactiveResilience4JCircuitBreakerFactory,然后通过构造方法传递给了StatePrinterGatewayFilter实例:

    package com.bolingcavalry.circuitbreakergateway.filter;

    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.circuitbreaker.resilience4j.ReactiveResilience4JCircuitBreakerFactory;
    import org.springframework.cloud.gateway.filter.GatewayFilter;
    import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
    import org.springframework.stereotype.Component;

    @Component
    public class StatePrinterGatewayFilterFactory extends AbstractGatewayFilterFactory
    {
    @Autowired
    ReactiveResilience4JCircuitBreakerFactory reactiveResilience4JCircuitBreakerFactory;

    @Override
    public String name() {
        return "CircuitBreakerStatePrinter";
    }
    
    @Override
    public GatewayFilter apply(Object config)
    {
        return new StatePrinterGatewayFilter(reactiveResilience4JCircuitBreakerFactory);
    }

    }

  • 最后是配置文件,完整的配置文件如下,可见我们将CircuitBreakerStatePrinter过滤器加了进来,放到最后:

    server:
    #服务端口
    port: 8081
    spring:
    application:
    name: circuitbreaker-gateway
    cloud:
    gateway:
    routes:
    - id: path_route
    uri: http://127.0.0.1:8082
    predicates:
    - Path=/hello/**
    filters:
    - name: CircuitBreaker
    args:
    name: myCircuitBreaker
    - name: CircuitBreakerStatePrinter

  • 再次运行单元测试类CircuitbreakerTest.java,如下图红框所示,断路器状态已经打印出来,至此,我们可以精确把握断路器的状态变化了:

  • 分析请求被filter漏掉的问题

    • 有个很明显的问题,聪明睿智的您当然不会忽略:上图绿框中的连续四个响应,对应的断路器状态都没有打印出来,要知道,咱们的过滤器可是要处理每一个请求的,怎么会连续漏掉四个呢?

    • 其实原因很容易推理出来:断路器CircuitBreaker的filter先执行,然后才是咱们的CircuitBreakerStatePrinter,而处于开启状态的断路器会直接返回错误给调用方,其后面的filter都不会执行了

    • 那么问题来了:如何控制CircuitBreaker和CircuitBreakerStatePrinter这两个filter的顺序,让CircuitBreakerStatePrinter先执行?

    • CircuitBreakerStatePrinter是咱们自己写的代码,修改StatePrinterGatewayFilter.getOrder的返回值可以调整顺序,但CircuitBreaker不是咱自己的代码呀,这可如何是好?

    • 老规矩,看看断路器的源码,前文已经分析过了,断路器最重要的代码是SpringCloudCircuitBreakerFilterFactory.apply方法,如下图红框,生成的filter是GatewayFilter接口的实现类:

    • 再看加载过滤器到集合的那段关键代码,在RouteDefinitionRouteLocator.loadGatewayFilters方法中,如下图所示,由于CircuitBreaker的filter并没有实现Ordered接口,因此执行的是红框中的代码,代表其顺序的值等于i+1,这个i就是遍历路由配置中所有过滤器时的一个从零开始的自增变量而已:

    • 回顾咱们的路由配置,CircuitBreaker在前,CircuitBreakerStatePrinter在后,所以,在添加CircuitBreaker的时候,i等于0,那么CircuitBreaker的order就等于i+1=1了

    • 而CircuitBreakerStatePrinter实现了Ordered接口,因此不会走红框中的代码,其order等于咱们写在代码中的值,咱们写的是10

    • 所以:CircuitBreaker的order等于1,CircuitBreakerStatePrinter等于10,当然是CircuitBreaker先执行了!

    再次修改

    • 知道了原因,改起来就容易了,我的做法很简单:StatePrinterGatewayFilter不再实现Ordered,这样就和CircuitBreaker的filter一样,执行的是上图红框中的代码,这样,在配置文件中,谁放在前面谁就先执行

    • 代码就不贴出来了,您自行删除StatePrinterGatewayFilter中和Ordered相关的部分即可

    • 配置文件调整后如下:

      server:
      #服务端口
      port: 8081
      spring:
      application:
      name: circuitbreaker-gateway
      cloud:
      gateway:
      routes:
      - id: path_route
      uri: http://127.0.0.1:8082
      predicates:
      - Path=/hello/**
      filters:
      - name: CircuitBreakerStatePrinter
      - name: CircuitBreaker
      args:
      name: myCircuitBreaker

    • 改完了,再次运行CircuitbreakerTest.java,如下图,这一次,每个请求都会打印出此时断路器的状态:

    知识点小结

    • 至此,用于观测断路器状态的自定义过滤器就算完成了,整个过程还是有不少知识点的,咱们来盘点一下:
    1. 常规的局部过滤器开发步骤
    2. 过滤器执行顺序的逻辑
    3. spring的依赖注入和自动装配
    4. 断路器的filter源码
    5. java的反射基本功

    你不孤单,欣宸原创一路相伴

    1. Java系列
    2. Spring系列
    3. Docker系列
    4. kubernetes系列
    5. 数据库+中间件系列
    6. DevOps系列

    欢迎关注公众号:程序员欣宸

    微信搜索「程序员欣宸」,我是欣宸,期待与您一同畅游Java世界…

    https://github.com/zq2599/blog_demos