Spring Cloud Gateway自定義過濾器實戰(觀測斷路器狀態變化)

歡迎訪問我的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：

5. 在配置文件中，添加您自定義的過濾器，該操作和之前的添加內置過濾器一模一樣

• 以上就是自定義過濾器的基本套路了，可見還是非常簡單的，接下來的實戰也是按照這個套路來的

• 在編寫自定義過濾器代碼之前，還有個攔路虎等着我們，也就是咱們過濾器的基本功能：如何取得斷路器的狀態

如何取得斷路器的狀態

• 前文的代碼分析中，咱們了解到斷路器的核心功能集中在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注解就能隨意使用了

• 至此，理論分析已全部完成，問題都已經解決，開始編碼

源碼下載

• 本篇實戰中的完整源碼可在GitHub下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：

名稱	鏈接	備注
項目主頁	https://github.com/zq2599/blog_demos	該項目在GitHub上的主頁
git倉庫地址(https)	https://github.com/zq2599/blog_demos.git	該項目源碼的倉庫地址，https協議
git倉庫地址(ssh)	[e-mail]: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<CircuitBreaker> seq = circuitBreakerRegistry.getAllCircuitBreakers();
            // 用名字過濾，myCircuitBreaker來自路由配置中
            circuitBreaker = seq.filter(breaker -> 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<Object>
{
    @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的反射基本功

• 本文與《Spring Cloud Gateway的斷路器(CircuitBreaker)功能》結合，誠意滿滿的帶給您理論結合實戰的體驗，希望能給您學習Spring Cloud Gateway的過程中帶來一些參考；

你不孤單，欣宸原創一路相伴

1. Java系列
2. Spring系列
3. Docker系列
4. kubernetes系列
5. 數據庫+中間件系列
6. DevOps系列

歡迎關注公眾號：程序員欣宸

微信搜索「程序員欣宸」，我是欣宸，期待與您一同暢游Java世界...
https://github.com/zq2599/blog_demos