SpringMvc 限流之 RateLimiter

概念

限流限流的目的是通过对并发访问/请求进行限速，或者对一个时间窗口内的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务、排队或等待、降级等处理

常用限流算法

常用的限流算法有两种：漏桶算法和令牌桶算法

漏桶算法思路很简单，水（请求）先进入到漏桶里，漏桶以一定的速度出水，当水流入速度过大会直接溢出，可以看出漏桶算法能强行限制数据的传输速率。
对于很多应用场景来说，除了要求能够限制数据的平均传输速率外，还要求允许某种程度的突发传输。这时候漏桶算法可能就不合适了，令牌桶算法更为适合。

令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。

控制并发数量

信号量Semaphore

Semaphore是一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可信号，做完自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。

简单的说：Semaphore(10)表示允许10个线程获取许可证，也就是最大并发数是10。

控制访问速率

限流工具类RateLimiter

Google开源工具包Guava提供了限流工具类RateLimiter，该类基于令牌桶算法来完成限流，非常易于使用。

RateLimiter源码分析

调用create接口时，实际实例化的为SmoothBursty类

  static final class SmoothBursty extends SmoothRateLimiter {/** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */final double maxBurstSeconds;/*** The currently stored permits.* 当前存储令牌数*/double storedPermits;/*** The maximum number of stored permits.* 最大存储令牌数*/double maxPermits;/*** The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits* per second has a stable interval of 200ms.* 添加令牌时间间隔，可以理解成生成一个令牌需要的时间，这里是微秒单位*/double stableIntervalMicros;.......}

RateLimiter 创建

public static RateLimiter create(double permitsPerSecond) {return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
}static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);rateLimiter.setRate(permitsPerSecond);return rateLimiter;
}

acquire()方法

public double acquire(int permits) {//计算获取这些请求需要让线程等待多长时间long microsToWait = reserve(permits);//让线程阻塞microTowait微秒长的时间stopwatch.sleepMicrosUninterruptibly(microsToWait);//返回阻塞的时间return 1.0 * microsToWait / SECONDS.toMicros(1L);}

reserve()方法

  final long reserve(int permits) {//检查permits是否合法checkPermits(permits);//保证线程安全synchronized (mutex()) {return reserveAndGetWaitLength(permits, stopwatch.readMicros());}}final long reserveAndGetWaitLength(int permits, long nowMicros) {long momentAvailable = reserveEarliestAvailable(permits, nowMicros);return max(momentAvailable - nowMicros, 0);}

reserveEarliestAvailable()

storedPermitsToSpend为桶中可以消费的令牌数，freshPermits为还需要的(需要补充的)令牌数，根据该值计算需要等待的时间，追加并更新到nextFreeTicketMicros


//获取requiredPermits个令牌，并返回需要等待到的时间点
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {resync(nowMicros);long returnValue = nextFreeTicketMicros;double storedPermitsToSpend = min(requiredPermits, this.storedPermits);double freshPermits = requiredPermits - storedPermitsToSpend;//当 requiredPermits>storedPermits才会有实际意义,这段代码允许我们提前获取令牌，但是这种情况会造成下一次令牌生成的时间推迟。有种预支工资的意思long waitMicros =storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)+ (long) (freshPermits * stableIntervalMicros);this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);//更新可消费的令牌this.storedPermits -= storedPermitsToSpend;return returnValue;}```#### resync()若当前时间晚于nextFreeTicketMicros，则计算该段时间内可以生成多少令牌，将生成的令牌加入令牌桶中并更新数据```void resync(long nowMicros) {// if nextFreeTicket is in the past, resync to nowif (nowMicros > nextFreeTicketMicros) {//时间间隔内生成的新令牌double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();//更新最大令牌数量storedPermits = min(maxPermits, storedPermits + newPermits);//下次可以获取的时间nextFreeTicketMicros = nowMicros;}}

tryAcquire函数可以尝试在timeout时间内获取令牌，如果可以则挂起等待相应时间并返回true，否则立即返回false
canAcquire用于判断timeout时间内是否可以获取令牌。

应用

拦截器配置，可以统一配置所有请求的上限，也可以单独对某个 url配置，该拦截器是基于 SpringMvc 的RequestMappingHandlerMapping获取url 进行操作。

<bean id="requestLimitInterceptor" class="cn.fraudmetrix.creditcloud.app.intercepters.RequestLimitInterceptor"><property name="globalRateLimiter" value="100" /><property name="urlProperties"><props><prop key="/creditcloud/test">100</prop></props></property></bean><!--拦截器配置--><mvc:interceptors><ref bean="requestLimitInterceptor" /></mvc:interceptors>

RequestLimitInterceptor 拦截器

public class RequestLimitInterceptor implements HandlerInterceptor ,BeanPostProcessor{private Logger logger = LoggerFactory.getLogger(RequestLimitInterceptor.class);private Integer globalRateLimiter = 100;private Map<PatternsRequestCondition, RateLimiter> urlRateMap;private Properties urlProperties;private UrlPathHelper urlPathHelper = new UrlPathHelper();@Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {if (urlRateMap != null) {String lookupPath = urlPathHelper.getLookupPathForRequest(request);for (PatternsRequestCondition patternsRequestCondition : urlRateMap.keySet()) {//使用spring DispatcherServlet的匹配器PatternsRequestCondition进行匹配List<String> matches = patternsRequestCondition.getMatchingPatterns(lookupPath);if (!matches.isEmpty()) {if (urlRateMap.get(patternsRequestCondition).tryAcquire(1000, TimeUnit.MILLISECONDS)) {logger.info(" 请求'{}'匹配到mathes {} ,成功获取令牌，进入请求。" ,lookupPath ,Joiner.on(",").join(patternsRequestCondition.getPatterns()) );} else {logger.info( " 请求'{}'匹配到mathes {}，超过限流速率，获取令牌失败。" ,lookupPath ,Joiner.on(",").join(patternsRequestCondition.getPatterns()));return false;}}}}return true;}@Overridepublic void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {}@Overridepublic void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {}/*** 限流的 URL与限流值的K/V 值** @param urlProperties*/public void setUrlProperties(Properties urlProperties) {this.urlProperties = urlProperties;}public void setGlobalRateLimiter(Integer globalRateLimiter) {this.globalRateLimiter = globalRateLimiter;}@Overridepublic Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {return bean;}@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {if(RequestMappingHandlerMapping.class.isAssignableFrom(bean.getClass())){if(urlRateMap==null){urlRateMap = new ConcurrentHashMap<>();}logger.info("we get all the controllers's methods and assign it to urlRateMap");RequestMappingHandlerMapping requestMappingHandlerMapping = (RequestMappingHandlerMapping)bean;Map<RequestMappingInfo, HandlerMethod> handlerMethods = requestMappingHandlerMapping.getHandlerMethods();for (RequestMappingInfo rmi : handlerMethods.keySet()) {PatternsRequestCondition pc = rmi.getPatternsCondition();urlRateMap.put(pc,RateLimiter.create(globalRateLimiter));}if(urlProperties!=null){for(String urlPatterns :urlProperties.stringPropertyNames()){String limit = urlProperties.getProperty(urlPatterns);if(!limit.matches("^-?\\d+$"))logger.error("the value {} for url patterns {} is not a number ,please check it ",limit,urlPatterns);urlRateMap.put(new PatternsRequestCondition(urlPatterns), RateLimiter.create(Integer.parseInt(limit)));}}}return bean;}
}

总结

RateLimiter通常用于限制访问某些物理或逻辑资源的速率。这与jdk并发包中的Semaphore相反，它限制并发访问的数量而不是速率(注意，并发和速率是密切相关的)。

参考：https://segmentfault.com/a/1190000012875897