限流神器之-Guava RateLimiter 实战

前段时间，项目中需要对某些访问量较高的路径进行访问并发数控制，以及有些功能，比如Excel导出下载功能，数据量很大的情况下，用户不断的点击下载按钮，重复请求数据库，导致线上数据库挂掉。于是在这样的情况下，这个限流组件应运而生，也许有人会提及SpringCloud zuul，其实它的现也是借助了RateLimiter。由于项目使用的是SpringBoot，也就没往外思考。反正最后功能实现了就行，毕竟殊途同归啊。本文只是用代码来快速的帮你理清整个限流的流程，至于RateLimiter中具体限流算法以及Semaphore信号量的具体实现还是得自己去深挖了，这里就不再展开了。正片时间到：

0、由于需要对限流的路径进行后台管理，那限流实体肯定是需要的

public class RateLimit {
    private String rateLimitId;
    /**
     * 限流路径，支持通配符，示例 /user/**
     */
    private String limitPath;
    /**
     * 每秒限流频率
     */
    private Integer permitsPerSecond;
    /**
     * 限流等待超时时间，单位s
     */
    private Integer permitsTimeOut;
    /**
     * 排序
     */private Integer orderNo;
    /**
     * 最大线程数
     */
    private Integer maxThread;
    /**
     * 创建时间
     */
    private Date gmtCreate;
   //get、set略
}

1、因为要借助RateLimiter类，所以再封装一个限流信息类

/**
 * @描述: 限流信息
 */
public class RateLimitInfo {

    private RateLimiter rateLimiter;

    private RateLimitVo rateLimitVo;

    private long lastUpdateTime;

    public RateLimitInfo(RateLimiter rateLimiter, RateLimitVo rateLimitVo, long lastUpdateTime) {
        this.rateLimiter = rateLimiter;
        this.rateLimitVo = rateLimitVo;
        this.lastUpdateTime = lastUpdateTime;
    }
    //get、set略
}

2、定义限流策略RateLimitStrategist

/**
 * @描述: 限流策略
 */
public class RateLimitStrategist {

    private PathMatcher pathMatcher = new AntPathMatcher();

    private final Map limiterMap = new LinkedHashMap<>();

    private final Map threadMap = new LinkedHashMap<>();

    /**
     * 更新频率，意为后台配置路径后5分钟生效
     */
    private static final long UPDATE_RATE = 1000*60*5;

    private long lastUpdateTime = 0;

    @Autowired
    private RateLimitManager rateLimitManager;

    public void init() {
        limiterMap.clear();
        threadMap.clear();
        List rateLimitVos = rateLimitManager.findListForPriority(); //查询数据库中配置的路径信息，需要自己实现
        if(CollectionUtils.isNotEmpty(rateLimitVos)) {
            return;
        }
        for (RateLimitVo rateLimitVo : rateLimitVos) {
            RateLimiter rateLimiter = RateLimiter.create(rateLimitVo.getPermitsPerSecond());
            limiterMap.put(rateLimitVo.getLimitPath(), new RateLimitInfo(rateLimiter, rateLimitVo, System.currentTimeMillis()));
            threadMap.put(rateLimitVo.getLimitPath(), new Semaphore(rateLimitVo.getMaxThread(), true));
        }
        lastUpdateTime = System.currentTimeMillis();
    }

    public boolean tryAcquire(String requestUri) {
        //目前设置5分钟更新一次
        if(System.currentTimeMillis() - lastUpdateTime > UPDATE_RATE) {
            synchronized (this) {
                if(System.currentTimeMillis() - lastUpdateTime > UPDATE_RATE) {
                    init();
                }
            }
        }

        for (Map.Entry entry : limiterMap.entrySet()) {
            if(!pathMatcher.match(entry.getKey(), requestUri)) {
                continue;
            }
            RateLimitInfo rateLimitInfo = entry.getValue();
            RateLimitVo rateLimitVo = rateLimitInfo.getRateLimitVo();
            RateLimiter rateLimiter = rateLimitInfo.getRateLimiter();
            boolean concurrentFlag = rateLimiter.tryAcquire(1, rateLimitVo.getPermitsTimeOut(), TimeUnit.SECONDS);
            if(!concurrentFlag) { //验证失败，直接返回
                return concurrentFlag;
            } else {
                if(threadMap.get(requestUri).availablePermits() != 0) { //当前路径对应剩余可执行线程数不为0
                    try {
                        //申请可执行线程
                        threadMap.get(requestUri).acquire();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return true;
                } else {
                    return false;
                }
            }
        }
        return true;
    }

    public void setLastUpdateTime(long lastUpdateTime) {
        this.lastUpdateTime = lastUpdateTime;
    }

    /**
     * 释放路径对应的线程数
     * @param requestURI
     */
    public void releaseSemaphore(String requestURI) {
        if(null != threadMap.get(requestURI)) {
            threadMap.get(requestURI).release();
        }
    }
}

3、定义拦截器RateLimitFilter，在拦截器中调用限流策略

/**
 * @描述: 限流过滤器，配置后生效
 */
public class RateLimitFilter implements Filter {

    private RateLimitStrategist rateLimitStrategist;

    private static final Logger LOGGER = LoggerFactory.getLogger(RateLimitFilter.class);

    @Override
    public void init(FilterConfig filterConfig) throws ServletException {

    }

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
        if(rateLimitStrategist == null) {
            rateLimitStrategist = InstanceFactory.getInstance(RateLimitStrategist.class);
        }
        HttpServletRequest req = (HttpServletRequest) request;
        HttpServletResponse res = (HttpServletResponse) response;
        String requestURI = req.getRequestURI();
        String contextPath = req.getContextPath();
        if(StringUtils.isNotBlank(contextPath)) {
            requestURI = StringUtils.substring(requestURI, contextPath.length());
        }
        if(!rateLimitStrategist.tryAcquire(requestURI)) {
            res.setContentType("text/html;charset=UTF-8");
            res.setStatus(HttpStatus.UNAUTHORIZED.value());
            response.getWriter().write("当前服务器繁忙，请稍后再试！");
            LOGGER.info(requestURI + "路径请求服务器繁忙，请稍后再试");
        } else {
            try {
                chain.doFilter(request, response);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                rateLimitStrategist.releaseSemaphore(requestURI);
            }
        }

    }

    @Override
    public void destroy() {

    }
}

4、需要的配置(采用注解也可以)

先在web.xml中引入过滤器（开始处）

    rateLimiter
    com.limit.filter.RateLimitFilter



    rateLimiter
    /*  

然后在context.xml中注入RateLimitStrategist