【基于spring-cloud-gateway实现本身的网关过滤器】

基于spring-cloud-gateway实现本身的网关过滤器

spring cloud gateway custom starter

自定义非阻塞式反应网关服务，集成鉴权、限流、响应的增强处置惩罚等等

• 环境要求
  1. <properties>
  2.       <java.version>17</java.version>
  3.       <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  4.       <maven.compiler.target>17</maven.compiler.target>
  5.       <maven.compiler.source>17</maven.compiler.source>
  6.       <java.source.version>17</java.source.version>
  7.       <java.target.version>17</java.target.version>
  8.       <spring-boot.version>3.1.12</spring-boot.version>
  9.       <spring-cloud.version>2022.0.5</spring-cloud.version>
  10.       <commons.pool2.version>2.12.0</commons.pool2.version>
  11.       <redisson.version>3.34.1</redisson.version>
  12.       <com.fastjson.jackson.version>2.17.2</com.fastjson.jackson.version>
  13.       <commons.lang3.version>3.16.0</commons.lang3.version>
  14.       <lombok.version>1.18.32</lombok.version>
  15.   </properties>

  复制代码
• GatewayFilter 路由过滤器
  
  
  
  • TokenFilterGatewayFilterFactory，鉴权处置惩罚
    
  • 设置示例：
    
  1. spring:
  2.   cloud:
  3.     gateway:
  4.       routes:
  5.         - id: testRoute
  6.           uri: http://127.0.0.1:8080
  7.           predicates:
  8.             - Path=/test/**
  9.           filters:
  10.             - name: TokenFilter
  11.               args:
  12.                 requestHeaderKey: auth

  复制代码
    
  
  
  
  • RlimterGatewayFilterFactory，自定义key，比如：我们想根据不同用户去做自定义限流，那么我们可以在本身的网关过滤工厂内里将limterKey设置为根据哀求头自定义的用户标识，来举行自定义的设置。相比重写spring-cloud-gateway内里自定义的keyResolver和RedisLimiter相对容易一些。
    
    
    
    • 设置示例：
      
    1. spring:
    2.     cloud:
    3.         gateway:
    4.           routes:
    5.             - id: route_test
    6.               uri: http://192.168.1.1:8091
    7.               predicates:
    8.                 - Path=/from/requestIds/to/appNames
    9.               filters:
    10.                 - name: RLimter
    11.                   args:
    12.                     limterKey: v1_10 #限流所需要的key
    13.                     rate: 5  #每秒允许的请求数
    14.                     crust: 0 #每秒令牌桶的填充数

    复制代码
       
    
    
    
    • 更多路由过滤器扩展中。。。
      
    
    
  自定义网关过滤工厂实现，TokenFilterGatewayFilterFactory和RlimterGatewayFilterFactory，以自定义限流器RlimterGatewayFilterFactory为例
  
  
  

• 起首定义网关过滤工厂功能接口，限流的key，速率和桶的大小，我们是按照spring-cloud-gateway内部限流的实现改编而来的，通过调用lua脚本，采用redis的令牌桶算法做限流
  

1. public interface RLimter {
3.     Mono<Response> isAllowed(String limitKey, String rate, String crust);
5.     @Setter
6.     @Getter
7.     @NoArgsConstructor
8.     @AllArgsConstructor
9.     class Response {
10.         private boolean allowed;
11.     }
12. }

复制代码

• 限流功能组件的注入和声明，注入我们必要的Bean，注入RedisScript，调用自定义的lua脚本，以及StringRedisTemplate，因为下面这段代码我是将整个网关作为的启动器，所以限流的实现类也一并交给spring管理了，RRedisRateLimiter
  

1. @Configuration
2. public class RLimterAutoConfiguration {
4.     @Bean(name = "rredisRequestRateLimiterScript")
5.     public RedisScript<?> redisRequestRateLimiterScript() {
6.         DefaultRedisScript redisScript = new DefaultRedisScript();
7.         redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("script/request_rate_limiter.lua")));
8.         redisScript.setResultType(List.class);
9.         return redisScript;
10.     }
12.     @Bean(name = "rredisRateLimiter")
13.     public RRedisRateLimiter redisRateLimiter(@Qualifier("rredisRequestRateLimiterScript") RedisScript<List<Long>> redisRequestRateLimiterScript, StringRedisTemplate redisTemplate) {
14.         return new RRedisRateLimiter(redisTemplate, redisRequestRateLimiterScript);
15.     }
17.     @Bean
18.     public RLimterGatewayFilterFactory rLimterGatewayFilterFactory(@Qualifier("rredisRateLimiter") RRedisRateLimiter redisRateLimiter) {
19.         return new RLimterGatewayFilterFactory(redisRateLimiter);
20.     }
21. }

复制代码

• 上面代码中提到的new ClassPathResource(“script/request_rate_limiter.lua”)，工程resources目录下即可，也是摘抄自spring-cloud-gateway内部的限流脚本，原封不动
  

1. redis.replicate_commands()
3. local tokens_key = KEYS[1]
4. local timestamp_key = KEYS[2]
5. -- redis.log(redis.LOG_WARNING, "tokens_key " .. tokens_key)
7. local rate = tonumber(ARGV[1])
8. local capacity = tonumber(ARGV[2])
9. local now = tonumber(ARGV[3])
10. local requested = tonumber(ARGV[4])
12. local fill_time = capacity / rate
13. local ttl = math.floor(fill_time * 2)
15. -- for testing, it should use redis system time in production
16. if now == nil then
17.   now = redis.call('TIME')[1]
18. end
20. --redis.log(redis.LOG_WARNING, "rate " .. ARGV[1])
21. --redis.log(redis.LOG_WARNING, "capacity " .. ARGV[2])
22. --redis.log(redis.LOG_WARNING, "now " .. now)
23. --redis.log(redis.LOG_WARNING, "requested " .. ARGV[4])
24. --redis.log(redis.LOG_WARNING, "filltime " .. fill_time)
25. --redis.log(redis.LOG_WARNING, "ttl " .. ttl)
27. local last_tokens = tonumber(redis.call("get", tokens_key))
28. if last_tokens == nil then
29.   last_tokens = capacity
30. end
31. --redis.log(redis.LOG_WARNING, "last_tokens " .. last_tokens)
33. local last_refreshed = tonumber(redis.call("get", timestamp_key))
34. if last_refreshed == nil then
35.   last_refreshed = 0
36. end
37. --redis.log(redis.LOG_WARNING, "last_refreshed " .. last_refreshed)
39. local delta = math.max(0, now-last_refreshed)
40. local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
41. local allowed = filled_tokens >= requested
42. local new_tokens = filled_tokens
43. local allowed_num = 0
44. if allowed then
45.   new_tokens = filled_tokens - requested
46.   allowed_num = 1
47. end
49. --redis.log(redis.LOG_WARNING, "delta " .. delta)
50. --redis.log(redis.LOG_WARNING, "filled_tokens " .. filled_tokens)
51. --redis.log(redis.LOG_WARNING, "allowed_num " .. allowed_num)
52. --redis.log(redis.LOG_WARNING, "new_tokens " .. new_tokens)
54. if ttl > 0 then
55.   redis.call("setex", tokens_key, ttl, new_tokens)
56.   redis.call("setex", timestamp_key, ttl, now)
57. end
59. -- return { allowed_num, new_tokens, capacity, filled_tokens, requested, new_tokens }
60. return { allowed_num, new_tokens }

复制代码

• 然后是限流实现类，通过传入上面lua脚本必要的四个参数即可，lua脚本中requested为每次从桶内里取出的令牌数，这个默认为1，此处不关注这个，默认值即可。
  

1. public class RRedisRateLimiter implements RLimter {
3.     private static final Logger logger = LoggerFactory.getLogger(RedisRateLimiter.class);
4.     private final StringRedisTemplate redisTemplate;
5.     private final RedisScript<List<Long>> script;
7.     public RRedisRateLimiter(StringRedisTemplate redisTemplate, RedisScript<List<Long>> script) {
8.         this.redisTemplate = redisTemplate;
9.         this.script = script;
10.     }
12.     static List<String> getKeys(String id) {
13.         String prefix = "request_rate_limiter.{" + id;
14.         String tokenKey = prefix + "}.tokens";
15.         String timestampKey = prefix + "}.timestamp";
16.         return Arrays.asList(tokenKey, timestampKey);
17.     }
19.     @Override
20.     public Mono<Response> isAllowed(String key, String rate, String crust) {
21.         List<String> keys = getKeys(key);
22.         List<Long> exec;
23.         try {
24.             exec = this.redisTemplate.execute(this.script, keys, rate, crust, "", "1");
25.         } catch (Throwable throwable) {
26.             logger.error("Error calling rate limiter lua", throwable);
27.             exec = Arrays.asList(1L, -1L);
28.         }
29.         assert exec != null;
30.         boolean allowed = exec.get(0) == 1L;
31.         return Mono.just(new Response(allowed));
32.     }
33. }

复制代码

• 最后就是定义我们本身的自定义限流网关工厂了，通过继承spring-cloud-gateway的一个父类，资助我们加载自定义的网关过滤工厂，AbstractGatewayFilterFactory，父类支持传入我们的自定义设置参数，Config，通过泛型参数定义本身的设置类，并在构造中传入。
  

1. public class RLimterGatewayFilterFactory extends AbstractGatewayFilterFactory<RLimterGatewayFilterFactory.Config2> {
3.     private final RRedisRateLimiter redisRateLimiter;
5.     public RLimterGatewayFilterFactory(RRedisRateLimiter redisRateLimiter) {
6.         super(Config2.class);
7.         this.redisRateLimiter = redisRateLimiter;
8.     }
10.     @Override
11.     public GatewayFilter apply(Config2 config) {
12.         return (exchange, chain) -> redisRateLimiter.isAllowed(config.limterKey, config.rate, config.crust).flatMap(response -> {
13.             if (response.isAllowed()) {
14.                 return chain.filter(exchange);
15.             } else {
16.                 ServerWebExchangeUtils.setResponseStatus(exchange, config.getStatusCode());
17.                 return exchange.getResponse().setComplete();
18.             }
19.         });
20.     }
22.     @Setter
23.     @Getter
24.     public static class Config2 implements HasRouteId {
25.         private String routeId;
26.         private String limterKey = "default";
27.         private String rate = "1";
28.         private String crust = "1";
29.         private HttpStatus statusCode = HttpStatus.TOO_MANY_REQUESTS;
31.         @Override
32.         public String getRouteId() {
33.             return routeId;
34.         }
36.         @Override
37.         public void setRouteId(String routeId) {
38.             this.routeId = routeId;
39.         }
40.     }
41. }

复制代码

• 这里我只是写了个简单的实例，假如实现自定义的限流，还必要在网关工厂内里，通过exchange拿到哀求体来举行相应的key的剖析，连合业务实现自定义的限流。设置可以参考一开始的yaml设置示例。可以测试当把令牌桶设置为0时，会给出TOO MANY REQUEST的429状态码。
  

全局过滤器案例

• 起首定义本身的上下文对象
  

1. @Setter
2. @Getter
3. public class RequestContext {
4.     private String requestId;
5.     private long requestStartTime;
6.     private String requestIp;
7. }

复制代码

• 全局过滤器实现
  

1. @Component
2. public class RequestContextFilter implements WebFilter, Ordered {
4.     private static final Logger logger = LoggerFactory.getLogger(RequestContextFilter.class);
6.     @Override
7.     public int getOrder() {
8.         return OrderConstant.REQUEST_CONTEXT_ORDER;
9.     }
11.     @Override
12.     public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
13.         long requestStartTime = SystemClock.now();
14.         String requestId = generateRequestId();
15.         ServerHttpRequest request = exchange.getRequest();
16.         String uri = request.getURI().getRawPath();
17.         String requestIp = IpUtils.tryGetRealIp(request);
18.         exchange.getResponse().getHeaders().add("requestId", requestId);
19.         logger.info("request start,requestId:{}, requestUri:{},ip:{},requestStartTime:{}", requestId, uri, requestIp, requestStartTime);
20.         RequestContext context = new RequestContext();
21.         context.setRequestId(requestId);
22.         context.setRequestStartTime(requestStartTime);
23.         context.setRequestIp(requestIp);
24.         return chain.filter(exchange).contextWrite(Context.of(RequestContext.class, context))
25.                 .doOnEach(signal -> {
26.                     long requestEndTime = SystemClock.now();
27.                     if (signal.isOnComplete()) {
28.                         logger.info("request end,requestId:{},response:{},requestEndTime:{},耗时ms:{}", requestId, exchange.getResponse().getStatusCode(), requestEndTime, (requestEndTime - requestStartTime));
29.                     }
30.                     if (signal.isOnError()) {
31.                         logger.info("request end,requestId:{},error:{},requestEndTime:{},耗时ms:{}", requestId, signal.getThrowable(), requestEndTime, (requestEndTime - requestStartTime));
32.                     }
33.                 });
34.     }
36.     private String generateRequestId() {
37.         return UUID.randomUUID().toString().replaceAll("-", "");
38.     }
39. }

复制代码