SpringBoot 限流实现

Posted 2022-11-30 高国藩

高并发访问时，缓存、限流、降级往往是系统的利剑，在互联网蓬勃发展的时期，经常会面临因用户暴涨导致的请求不可用的情况，甚至引发连锁反映导致整个系统崩溃。这个时候常见的解决方案之一就是限流了，当请求达到一定的并发数或速率，就进行等待、排队、降级、拒绝服务等

限流算法介绍
a、令牌桶算法
令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。 当桶满时，新添加的令牌被丢弃或拒绝。

b、漏桶算法
其主要目的是控制数据注入到网络的速率，平滑网络上的突发流量，数据可以以任意速度流入到漏桶中。 漏桶算法提供了一种机制，通过它，突发流量可以被整形以便为网络提供一个稳定的流量。 漏桶可以看作是一个带有常量服务时间的单服务器队列，如果漏桶为空，则不需要流出水滴，如果漏桶（包缓存）溢出，那么水滴会被溢出丢弃

c、计算器限流
计数器限流算法是比较常用一种的限流方案也是最为粗暴直接的，主要用来限制总并发数，比如数据库连接池大小、线程池大小、接口访问并发数等都是使用计数器算法

如：使用 AomicInteger 来进行统计当前正在并发执行的次数，如果超过域值就直接拒绝请求，提示系统繁忙

限流具体代码实践

• 导入mvn依赖包

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>21.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
</dependencies>

• 实现redis配置信息

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.io.Serializable;

/**
 * redis配置
 */
@Configuration
public class RedisConfig 

    @Bean
    public RedisTemplate<String, Serializable> limitRedisTemplate(LettuceConnectionFactory redisConnectionFactory) 
        RedisTemplate<String, Serializable> template = new RedisTemplate<>();
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.setConnectionFactory(redisConnectionFactory);
        return template;
    

当然，你也可以使用普通的jedis.eval方法执行下面的lua脚本命令。

    /**
     * 执行lua脚本命令
    * @author 高国藩
    * @date 2019年6月19日 下午5:48:37
    * @param scriptKey
    * @param keys
    * @param args
    * @return
     */
    public Object eval(String luaScript, ImmutableList<String> keys, String... args) 
        Jedis jedis = null;
        boolean isBroken = false;
        try 
            jedis = this.getJedis();
            return jedis.eval(luaScript, keys, Arrays.asList(args));
        
        catch (Exception e) 
            isBroken = true;
        
        finally 
            release(jedis, isBroken);
        
        return null;
    

• 实现限流注解Class

package com.soul.home.lws.request.conf;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * 限流注解定义
* @author 高国藩
* @date 2019年6月19日 下午5:09:53
 */
@Target( ElementType.METHOD, ElementType.TYPE )
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Limit 

    /** 资源的名字 */
    String name() default "";

    /** 资源的key */
    String key() default "";

    /** Key的prefix */
    String prefix() default "";

    /** 给定的时间段 单位秒 */
    int period();

    /** 最多的访问限制次数 */
    int count();

    /**
     * 限流类型
    * @author 高国藩
    * @date 2019年6月19日 下午5:10:09
    * @return
     */
    LimitType limitType() default LimitType.IP;

package com.soul.home.lws.request.conf;

/**
 * 限流分类枚举
* @author 高国藩
* @date 2019年6月19日 下午5:10:41
 */
public enum LimitType 
    
    /** 自定义key */
    CUSTOMER,
    
    /** 根据请求者IP */
    IP;

• 实现aop拦截器信息

我们可以通过编写 Lua 脚本实现自己的API，核心就是调用 execute 方法传入我们的 Lua 脚本内容，然后通过返回值判断是否超出我们预期的范围，超出则给出错误提示。

package com.soul.home.lws.request.conf;

import com.google.common.collect.ImmutableList;
import com.soul.home.lws.system.exception.RequestLimitException;
import com.soul.home.lws.system.service.RedisService;

import net.sf.json.JSONObject;

import org.apache.commons.lang3.StringUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

/**
 * 
* @author 高国藩
* @date 2019年6月20日 上午11:48:06
 */
@Aspect
@Configuration
public class LimitInterceptor 

    private static final Logger logger = LoggerFactory.getLogger(LimitInterceptor.class);
    
    @Autowired RedisService redisService;

    @Around("execution(public * *(..)) && @annotation(com.soul.home.lws.request.conf.Limit)")
    public Object interceptor(ProceedingJoinPoint pjp) 
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();
        Limit limitAnnotation = method.getAnnotation(Limit.class);
        LimitType limitType = limitAnnotation.limitType();
        String name = limitAnnotation.name();
        String key;
        Integer limitPeriod = limitAnnotation.period();
        Integer limitCount = limitAnnotation.count();
        switch (limitType) 
            case IP:
                key = getIpAddress();
                break;
            case CUSTOMER:
                key = limitAnnotation.key();
                break;
            default:
                key = StringUtils.upperCase(method.getName());
        
        ImmutableList<String> keys = ImmutableList.of(StringUtils.join(limitAnnotation.prefix(), key));
        try 
            String luaScript = buildLuaScript();
            
            Object resultCount = redisService.eval(luaScript, keys, limitCount.toString(), limitPeriod.toString());
            Number count = Integer.valueOf(resultCount.toString());
            logger.info("Access try count is  for name= and key = ", count, name, key);
            if (count != null && count.intValue() <= limitCount) 
                return pjp.proceed();
             else 
                sendLimtErrorCodeMessage();
                return null;
            
         catch (Throwable e) 
            if (e instanceof RuntimeException) 
                throw new RuntimeException(e.getLocalizedMessage());
            
            throw new RuntimeException("server exception");
        
    
    
    
    /**
     * 抛出请求异常信息，提示请求客户端
    * @author 高国藩
    * @date 2019年6月20日 上午11:47:42
     */
    private void sendLimtErrorCodeMessage() 
        try 
            ServletRequestAttributes servletRequestAttributes = (ServletRequestAttributes)RequestContextHolder.getRequestAttributes();
            HttpServletRequest request = servletRequestAttributes.getRequest();
            HttpServletResponse response = servletRequestAttributes.getResponse();
            if (request == null) 
                throw new RequestLimitException("方法中缺失HttpServletRequest参数");
            
            response.setCharacterEncoding("UTF-8");  
            response.setContentType("application/json; charset=utf-8");
            Map<String, Object> error = new HashMap<>();
            error.put("ret", -1);
            error.put("errCode", 100);
            error.put("errMsg", "请求频繁，请稍后重试");
            String str = JSONObject.fromObject(error).toString();
            byte[] bytes = str.getBytes();
            response.setContentLength(bytes.length);
            ServletOutputStream outputStream = response.getOutputStream();
            outputStream.write(bytes);
            outputStream.flush();
         catch (Exception e) 
            // TODO: handle exception
        
    
    

    /**
     * 限流LUA脚本命令
    * @author 高国藩
    * @date 2019年6月20日 上午11:47:32
    * @return
     */
    public String buildLuaScript() 
        StringBuilder lua = new StringBuilder();
        lua.append("local c");
        lua.append("\\nc = redis.call('get',KEYS[1])");
        // 调用不超过最大值，则直接返回
        lua.append("\\nif c and tonumber(c) > tonumber(ARGV[1]) then");
        lua.append("\\nreturn c;");
        lua.append("\\nend");
        // 执行计算器自加
        lua.append("\\nc = redis.call('incr',KEYS[1])");
        lua.append("\\nif tonumber(c) == 1 then");
        // 从第一次调用开始限流，设置对应键值的过期
        lua.append("\\nredis.call('expire',KEYS[1],ARGV[2])");
        lua.append("\\nend");
        lua.append("\\nreturn c;");
        return lua.toString();
    

    private static final String UNKNOWN = "unknown";
    

    /**
     * 动态获取ip地址信息(存在反向代理信息和正向代理信息)
    * @author 高国藩
    * @date 2019年6月20日 上午11:48:22
    * @return
     */
    public String getIpAddress() 
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        String ip = request.getHeader("x-forwarded-for");
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) 
            ip = request.getHeader("Proxy-Client-IP");
        
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) 
            ip = request.getHeader("WL-Proxy-Client-IP");
        
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) 
            ip = request.getRemoteAddr();
        
        return ip;
    

• 实现最终限流

import com.carry.annotation.Limit;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.atomic.AtomicInteger;


@RestController
public class LimiterController 

    private static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger();

    @Limit(key = "test", period = 100, count = 10, name="resource", prefix = "limit")
    @GetMapping("/test")
    public int testLimiter() 
        // 意味着100S内最多可以访问10次
        return ATOMIC_INTEGER.incrementAndGet();
    

注意：上面例子保存在redis中的key值应该为“limittest”，即@Limit中prefix的值+key的值