前段时间,项目中需要对某些访问量较高的路径进行访问并发数控制,以及有些功能,比如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<String, RateLimitInfo> limiterMap = new LinkedHashMap<>(); private final Map<String, Semaphore> 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<RateLimitVo> 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<String, RateLimitInfo> 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中引入过滤器(开始处)
<filter>
<filter-name>rateLimiter</filter-name>
<filter-class>com.limit.filter.RateLimitFilter</filter-class>
</filter> <filter-mapping>
<filter-name>rateLimiter</filter-name>
<url-pattern>/*</url-pattern>
</filter-mapping>
然后在context.xml中注入RateLimitStrategist
<bean id="rateLimitStrategist" class="com.limit.factory.RateLimitStrategist" />
5、代码tag-0.0.1是项目单节点部署可用,版本tag-0.0.2为适应多节点部署改为redis来实现对处理线程数的控制
**需要源码留邮箱**