spring cloud gateway 实现接口限流

逃离我推掉我的手 2022-01-11 02:53 401阅读 0赞

限流一般有两个实现方式，令牌桶和漏桶

令牌桶是初始化令牌（容器）的个数，通过拿走里边的令牌就能通过， 没有令牌不能报错，可以设置向容器中增加令牌的速度和最大个数

漏桶是向里边放入请求，当请求数量达到最大值后，丢弃，漏桶中的数据以一定速度流出，没有则不流出

令牌桶实现方式如下：  
pom.xml

<dependency>
    	 <groupId>com.github.vladimir-bukhtoyarov</groupId>
    	 <artifactId>bucket4j-core</artifactId>
    	 <version>4.0.0</version>
    </dependency>

创建下边类并且继承下边类

package com.gla.datacenter.filter;
    import io.github.bucket4j.Bandwidth;
    import io.github.bucket4j.Bucket;
    import io.github.bucket4j.Bucket4j;
    import io.github.bucket4j.Refill;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    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.time.Duration;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * @Description: 自定义过滤器进行限流
     */
    @Data
    public class GatewayRateLimitFilterByIP implements GatewayFilter, Ordered {
    	 private final Logger log = LoggerFactory.getLogger(GatewayRateLimitFilterByIP.class);
    	 /**
    	 * 单机网关限流用一个ConcurrentHashMap来存储 bucket，
    	 * 如果是分布式集群限流的话，可以采用 Redis等分布式解决方案
    	 */
    	 private static final Map<String, Bucket> LOCAL_CACHE = new ConcurrentHashMap<>();
    	 /**
    	 * 桶的最大容量，即能装载 Token 的最大数量
    	 */
    	 int capacity;
    	 /**
    	 * 每次 Token 补充量
    	 */
    	 int refillTokens;
    	 /**
    	 *补充 Token 的时间间隔
    	 */
    	 Duration refillDuration;
    	 
    	public GatewayRateLimitFilterByIP() {}
    	
    	 /**
    	 *
    	 * @param capacity 即能装载 Token 的最大数量.
    	 * @param refillTokens
    	 * @param refillDuration
    	 */
    	 public GatewayRateLimitFilterByIP(int capacity, int refillTokens, Duration refillDuration) {
    		 this.capacity = capacity;
    		 this.refillTokens = refillTokens;
    		 this.refillDuration = refillDuration;
    	 }
    	 
    	 private Bucket createNewBucket() {
    		 Refill refill = Refill.of(refillTokens, refillDuration);
    		 Bandwidth limit = Bandwidth.classic(capacity, refill);
    		 return Bucket4j.builder().addLimit(limit).build();
    	 }
    	 
    	 @Override
    	 public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    		 String ip = exchange.getRequest().getRemoteAddress().getAddress().getHostAddress();
    		 //若ip不存在则创建一个Bucket（令牌桶）
    		 Bucket bucket = LOCAL_CACHE.computeIfAbsent(ip, k -> createNewBucket());
    		 log.info("IP:{} ,令牌通可用的Token数量:{} " ,ip,bucket.getAvailableTokens());
    		 if (bucket.tryConsume(1)) {
    			 return chain.filter(exchange);
    		 } else {
    			 //当可用的令牌书为0是，进行限流返回429状态码
    			 log.error("IP:{} ,限制访问:{} " ,ip,bucket.getAvailableTokens());
    			 exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
    			 return exchange.getResponse().setComplete();
    		 }
    	 }
    	 
    	 @Override
    	 public int getOrder() {
    		return -1000;
    	 }
    	 
    	 public static Map<String, Bucket> getLocalCache() {
    		return LOCAL_CACHE;
    	 }
    	 
    
    }

配置路由

@Slf4j
    @Configuration
    public class Roulauration {
    
    @Bean
    	public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    		//生成比当前时间早一个小时的UTC时间
    		ZonedDateTime minusTime = LocalDateTime.now().minusHours(1).atZone(ZoneId.systemDefault());
    		return builder.routes()
    				.route(r ->r.path("/demo/**")
    						//过滤器
    						.filters(f -> f.filter(new APIGatewayFilter())
    								.filter(new GatewayRateLimitFilterByIP(10,1, Duration.ofSeconds(1))))
    						.uri("http://192.168.26.113:8001/demo").order(0).id("demo_route"))
    				.route(r ->r.path("/test")
    								.uri("http://192.168.26.113/system/nav/login").id("jd_route")
    				).build();
    				}
    }