限流——令牌桶算法

令牌桶简单来说就是有一个桶，然后假设里面存放了1000个令牌，我们访问一个接口需要有一个令牌，然后令牌桶中就会减少1个令牌，所以最多只能有1000个请求拿到令牌，但是，我们一般不会只有1000个令牌，所以，我们还需要设置每隔一段时间就会自动生成令牌

然后我们使用lua+redis来进行令牌桶限流代码的编写
这里使用的是springboot，先上代码

TokenBucket.class

@Slf4j
public class TokenBucket {


    //当前令牌桶的key
    private String key;

    //令牌桶的最大容量
    private int maxTokens;

    //每秒产生令牌的数量
    private int secTokens;

    private StringRedisTemplate redisTemplate;

    public TokenBucket(String key, int maxTokens, int secTokens){
        this.key=key;
        this.maxTokens=maxTokens;
        this.secTokens=secTokens;
        //手动从容器中获取Redis模板对象
        this.redisTemplate= SpringUtils.getBean(StringRedisTemplate.class);
        init();
    }

    //初始化令牌桶
    private void init(){
        log.info("###################开始进行令牌桶初始化#######################");
        //初始化token lua脚本
        Properties info = redisTemplate.getConnectionFactory().getConnection().info();
        String server = info.getProperty("server");
        //执行lua脚本
        redisTemplate.execute(new DefaultRedisscript(TokenLua.initBucket),
                Collections.singletonList(String.valueOf(this.key)),
                String.valueOf(this.maxTokens),
                String.valueOf(this.secTokens),
                TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis())+"");
		
			log.info("###################令牌桶初始完成#######################");
    }


    
    public double getTokens(int tokens)  {
        long waitTime= (long) redisTemplate.execute(new DefaultRedisscript(TokenLua.getToken,Long.class),
                Collections.singletonList(String.valueOf(key)),
                String.valueOf(tokens+""));

        if (waitTime>0){
            try {
                Thread.sleep(waitTime/1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return waitTime;
    }


    
    public boolean getTokens(int tokens,int timout,TimeUnit unit) {
        long waitTime= (long) redisTemplate.execute(new DefaultRedisscript(TokenLua.getToken,Long.class),
                Collections.singletonList(String.valueOf(key+"")),
                String.valueOf(tokens+""),
                unit.toMicros(timout)+"");

        if (waitTime==-1){
            return false;
        }
        if (waitTime>0){
            try {
                Thread.sleep(waitTime/1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return true;
    }
    
    public boolean getTokensNow(int tokens) {
        return getTokens(tokens,0,TimeUnit.MICROSECONDS);
    }
}

SpringUtils.class

//用来从IOC容器获取一个Bean
@Component
public class SpringUtils implements BeanFactoryAware {

    private static BeanFactory beanFactory;

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        SpringUtils.beanFactory=beanFactory;
    }


    public static  T getBean(Class c){
        return (T) beanFactory.getBean(c);
    }


    public static  T getBean(String beanName){
        return (T) beanFactory.getBean(beanName);
    }

}

使用方法，在这里以我们的gateway网关为例

@Component
public class TokenLimitFilter implements GatewayFilter {

    
    private Map tongMap=new ConcurrentHashMap();

    @Autowired
    private BucketConfig bucketConfig;

    @Override
    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        //令牌桶限流---URL限流
        ServerHttpRequest request = exchange.getRequest();
        String requestPath = request.getPath().value();

        Map map = bucketConfig.getMap();
        BucketConfig.Bucket bucket = map.get(requestPath);

        TokenBucket tokenBucket = tongMap.computeIfAbsent(requestPath,
                s->  new TokenBucket(requestPath,bucket.getMaxToken(),bucket.getSecToken()));

        boolean flag = tokenBucket.getTokensNow(2);

        if (flag){
            System.out.println("拿到令牌");
            //请求放行
            return chain.filter(exchange);
        }
        //没有拿到令牌，直接返回服务器繁忙
        Result resultData = Result.fail(ResultCode.SERVER_BUSY);

        ServerHttpResponse response = exchange.getResponse();
        response.getHeaders().put("Content-Type", Collections.singletonList("application/json?charset=utf-8"));
        response.getHeaders().put("Access-Control-Allow-Orgin", Collections.singletonList("*"));
        DataBuffer dataBuffer = null;
        try {
            dataBuffer = response.bufferFactory().wrap(JSON.toJSONString(resultData).getBytes("utf-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        Mono mono = response.writeWith(Mono.just(dataBuffer));
        System.out.println("没有拿到令牌");
        return mono;

    }

    public Map getTongMap() {
        return tongMap;
    }
}

之后只要在执行请求的时候，接入这个拦截器就行了

@Component
public class TokenLimitFilterFactory extends AbstractGatewayFilterFactory {

    @Autowired
    private TokenLimitFilter tokenLimitFilter;

    @Override
    public GatewayFilter apply(Object config) {
        return tokenLimitFilter;
    }

    @Override
    public String name() {
        return "Token_Limiter";
    }
}

配置

spring:
  cloud:
    gateway:
      discovery:
        locator:
          enabled: true
      routes:
        - id: test_route
          uri: http://www.baidu.com
          predicates:
            - Query=url,test
        - id: product_route
          uri: lb://mall-product
          predicates:
            - Path=/product/**
          filters:
            - RewritePath=/product/(?.*),/${segment}
            - Token_Limiter