Java服務限流算法
作者：程序猿Clover
其實這幾種算法，不能說哪一個是最好的，只能說是要的業務邏輯是什么樣的，選擇合適的限流算法來滿足自己的業務實現，沒有最優，只有最合適。
一、概述
限流其實就是對服務的請求做一下QPS的控制，對于有些免登錄的接口需要做一下訪問的限制，不能無限制的去請求接口，不然的話會給服務器造成很大的壓力，而且我們也希望一些接口做一下控制，控制請求量，這樣我們就可以做一個plugin對服務做限流操作，超出限流就返回請求失敗，保證系統的穩定運行。主要概念就是閾值以及拒絕策略，實際中需要用到限流的的比如，驗證碼，白名單，當然也有容器的限流，比如nginx就是比較常用的，可以做一下簡單的處理。
二、限流算法類型
幾種算法的使用，一些基礎代碼如下
限流代碼基礎類
@RequestLimiter
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Order(Ordered.HIGHEST_PRECEDENCE)
public @interface RequestLimiter {
/**
* 限流類型 ，具體見枚舉類 RequestLimitType
*/
RequestLimitType type() default RequestLimitType.TOKEN;
/**
* 限流訪問數
*/
int limitCount() default 100;
/**
* 限流時間段
*/
long time() default 60;
/**
* 限流時間段 時間單位
*/
TimeUnit unit() default TimeUnit.SECONDS;
/**
* 漏出或者生成令牌時間間隔，單位 毫秒 (當type為TOKEN、LEAKY_BUCKET時生效)
*/
long period() default 1000;
/**
* 每次生成令牌數或者漏出水滴數 (當type為TOKEN、LEAKY_BUCKET時生效)
*/
int limitPeriodCount() default 10;
}
LimitKeyConstant
public class LimitKeyConstant {
/**
* 令牌桶鍵名
*/
public static final String QPS_TOKEN = "request:limit:qps:tokenBucket:";
/**
* 漏桶鍵名
*/
public static final String QPS_LEAKY_BUCKET = "request:limit:qps:leakyBucket:";
/**
* 固定窗口鍵名
*/
public static final String QPS_FIXED_WINDOW = "request:limit:qps:fixedWindow:";
/**
* 滑動窗口鍵名
*/
public static final String QPS_SLIDE_WINDOW = "request:limit:qps:slideWindow:";
}
RequestLimitType
public enum RequestLimitType {
/**
* 令牌算法
*/
TOKEN(1, "令牌算法"),
/**
* 漏桶算法
*/
LEAKY_BUCKET(2, "漏桶算法"),
/**
* 固定窗口
*/
FIXED_WINDOW(3, "固定窗口"),
/**
* 滑動窗口
*/
SLIDE_WINDOW(4, "滑動窗口");
private Integer type;
private String desc;
RequestLimitType(Integer type, String desc) {
this.type = type;
this.desc = desc;
}
public Integer getType() {
return type;
}
public String getDesc() {
return desc;
}
}
RequestLimitAspect
@Slf4j
@Aspect
@Component
public class RequestLimitAspect {
@Autowired
private RequestLimitFactory factory;
/**
* 切入點
*/
@Pointcut(value = "@annotation(com.common.limit.annotation.RequestLimiter)")
public void requestLimit(){
// 切入點方法
}
/**
* 前置切點
*
* @param
@Before("requestLimit()")
public void doBefore(JoinPoint joinPoint){
RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
HttpServletRequest request = (HttpServletRequest) requestAttributes.resolveReference(RequestAttributes.REFERENCE_REQUEST);
Signature signature = joinPoint.getSignature();
MethodSignature methodSignature = (MethodSignature) signature;
Method targetMethod = methodSignature.getMethod();
RequestLimiter limiter = targetMethod.getAnnotation(RequestLimiter.class);
RequestLimitService service = factory.build(limiter.type());
if (service != null) {
RequestLimitParam param = new RequestLimitParam();
param.setLimiter(limiter);
param.setKey(signature.getName());
if (service.checkRequestLimit(param)) {
throw new LimitException("請求過于頻繁，請稍后再重試！");
}
}
}
}
RequestLimitFactory
@Slf4j
@Component
public class RequestLimitFactory implements ApplicationContextAware {
private static final Map<RequestLimitType, RequestLimitService> MAP = new ConcurrentHashMap<>();
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
try {
applicationContext.getBeansOfType(RequestLimitService.class).values().forEach(service -> MAP.put(service.getType(), service));
} catch (Exception e) {
log.error("初始化限流策略異常", e);
}
}
/**
* 構建service
*
* @param type 限流類型
* @return
public RequestLimitService build(RequestLimitType type){
return MAP.get(type);
}
}
RequestLimitService
public interface RequestLimitService {
/**
* 檢測是否限流
*
* @param param 限流參數
* @return
boolean checkRequestLimit(RequestLimitParam param);
/**
* 獲取當前限流類型
*
* @return
RequestLimitType getType();
/**
* 獲取帶注解方法列表
*
* @param resourcePatternResolver 資源查詢
* @param limitType 注解類型
* @param scanPackage 掃描包路徑
* @return
default List<RequestLimitParam> getTokenLimitList(ResourcePatternResolver resourcePatternResolver, RequestLimitType limitType,
String scanPackage){
try {
List<RequestLimitParam> list = new ArrayList<>();
Resource[] resources = resourcePatternResolver.getResources(ResourcePatternResolver.CLASSPATH_ALL_URL_PREFIX + scanPackage +
"/**/*.class");
MetadataReaderFactory metaReader = new CachingMetadataReaderFactory();
for (Resource resource : resources) {
MetadataReader reader = metaReader.getMetadataReader(resource);
AnnotationMetadata annotationMetadata = reader.getAnnotationMetadata();
Set<MethodMetadata> annotatedMethods = annotationMetadata.getAnnotatedMethods(RequestLimiter.class.getCanonicalName());
annotatedMethods.forEach(methodMetadata -> {
RequestLimiter limiter = methodMetadata.getAnnotations().get(RequestLimiter.class).synthesize();
if (!limitType.equals(limiter.type())) {
return;
}
RequestLimitParam param = new RequestLimitParam();
param.setKey(methodMetadata.getMethodName());
param.setLimiter(limiter);
list.add(param);
});
}
return list;
} catch (IOException e) {
return Collections.emptyList();
}
}
}
固定時間窗口算法
圖解
介紹
其實就是原子計數法，就是在固定時間內，允許請求量是多少，每次請求就在計數器上加1，設置計數器的過期時間，當計數器的閾值達到限流配置的數時候，就執行拒絕策略，超過了時間，計數器就會重新歸0。
比如上圖中，會限制在每秒限制請求數為2，就是在每秒的時間會限制請求為2，但是會出現極端的情況，比如在前一個時間段中的前500ms和后500ms，請求數都是2，這樣就會看到在這一秒內是有4個請求的，這就是會出現請求的問題，當然這也是最簡單的限流算法。
代碼
@Slf4j
@Service
public class FixedWindowRateLimitServiceImpl implements RequestLimitService {
@Autowired
private RedisConnectionFactory factory;
@Override
public boolean checkRequestLimit(RequestLimitParam param){
String key = LimitKeyConstant.QPS_FIXED_WINDOW + param.getKey();
RequestLimiter limiter = param.getLimiter();
RedisAtomicInteger atomicCount = new RedisAtomicInteger(key, factory);
int count = atomicCount.getAndIncrement();
if (count == 0) {
atomicCount.expire(limiter.time(), limiter.unit());
}
log.info("FixedWindowRateLimitServiceImpl time：{} unit：{} allow visit {} ", limiter.time(), limiter.unit(), limiter.limitCount());
// 檢測是否到達限流值
if (count >= limiter.limitCount()) {
log.info("FixedWindowRateLimitServiceImpl limit controller key:{},time:{},name:{} to visit :{}", key, limiter.time(),
limiter.unit().name(), limiter.limitCount());
return true;
} else {
return false;
}
}
@Override
public RequestLimitType getType(){
return RequestLimitType.FIXED_WINDOW;
}
}
滑動時間窗口算法
圖解
介紹
滑動時間窗口算法，其實就是對固定窗口的改進，知道了固定時間窗口會出現極端的情況，那滑動就在下一個臨界的時候，進行處理時間，其實就是在某一段時間進行處理時間。
比如上圖中每 500ms 滑動一次窗口，可以發現窗口滑動的間隔越短，時間窗口的臨界突變問題發生的概率也就越小，不過只要有時間窗口的存在，還是有可能發生時間窗口的臨界突變問題。
這個是記錄下所有的請求時間點，新請求先判斷最近指定時間范圍內的請求數量是否超過指定閾值，來確定是否達到限流，雖然沒有時間窗口突變的問題，限流比較準確，但是要記錄下每次請求的時間點，所以占用的內存較多。
代碼
@Slf4j
@Service
public class SlideWindowRateLimitServiceImpl implements RequestLimitService {
@Autowired
private RedisService redisService;
@Override
public boolean checkRequestLimit(RequestLimitParam param){
String key = LimitKeyConstant.QPS_SLIDE_WINDOW + param.getKey();
RequestLimiter limiter = param.getLimiter();
long current = System.currentTimeMillis();
long duringTime = limiter.unit().toMillis(limiter.time());
Long count = redisService.setCount(key, current - duringTime, current);
// 清除有效期外的數據
redisService.setRemoveRangeByScore(key, 0, current - duringTime - 1f);
log.info("SlideWindowRateLimitServiceImpl time：{} unit：{} allow visit {}", limiter.time(), limiter.unit(), limiter.limitCount());
// 檢測是否到達限流值
if (count != null && count >= limiter.limitCount()) {
log.info("SlideWindowRateLimitServiceImpl limit controller key:{},time:{},name:{} to visit :{}", key, limiter.time(),
limiter.unit().name(), limiter.limitCount());
return true;
} else {
redisService.setAdd(key, UUID.randomUUID().toString(), current);
return false;
}
}
@Override
public RequestLimitType getType(){
return RequestLimitType.SLIDE_WINDOW;
}
}
漏桶算法
圖解
介紹
此算法就是定義一個桶的容量，然后每次的請求過來都放在桶里面，一直等到桶滿了以后就會執行拒絕策略，然后在桶不滿的情況下，會按照固定的速率去執行請求，其實就是按照固定流速去執行請求，保證單位時間內的執行請求量是固定的。
漏桶就是按照某一個請求的穩定的速度處理發來的請求數量，可以很好地保證系統的穩定運行，只能平穩處理請求，這也是他的一個缺點，不能處理面對突然來的高的請求量，會導致請求一直處于哎隊列等待中，不能面對高并發下的請求處理，比較保守的處理邏輯
代碼
@Slf4j
@Service
public class LeakyBucketRateLimitServiceImpl implements RequestLimitService {
@Autowired
private ResourcePatternResolver resourcePatternResolver;
@Autowired
private RedisService redisService;
@Resource(name = Constants.THREAD_POOL_TASK_BEAN_NAME)
private ThreadPoolTaskScheduler executor;
@Value("${limit.scan.package}")
private String scanPackage;
@Override
public boolean checkRequestLimit(RequestLimitParam requestLimitParam) {
String key = LimitKeyConstant.QPS_LEAKY_BUCKET + requestLimitParam.getKey();
Long size = redisService.listSize(key);
if (size != null && size >= requestLimitParam.getLimiter().limitCount()) {
log.info("LeakyBucketRateLimitServiceImpl limit key：{}", requestLimitParam.getKey());
return true;
} else {
log.info("LeakyBucketRateLimitServiceImpl not full，limit key：{} ，current size：{}，total size：{}", requestLimitParam.getKey(),
size, requestLimitParam.getLimiter().limitCount());
redisService.listLeftPush(key, UUID.randomUUID().toString());
return false;
}
}
/**
* 定數流出令牌
*/
@PostConstruct
public void init() {
List<RequestLimitParam> list = this.getTokenLimitList(resourcePatternResolver, RequestLimitType.LEAKY_BUCKET, scanPackage);
if (list.isEmpty()) {
log.info("LeakyBucketRateLimitServiceImpl annotation is empty,end current task pool");
return;
}
list.forEach(requestLimitDTO -> {
executor.scheduleAtFixedRate(() -> {
String key = LimitKeyConstant.QPS_LEAKY_BUCKET + requestLimitDTO.getKey();
//截取List在start和end之間的元素處key列表
redisService.listTrim(key, requestLimitDTO.getLimiter().limitPeriodCount(), -1);
log.info("LeakyBucketRateLimitServiceImpl limit key:{},limitPeriodCount:{}", key,
requestLimitDTO.getLimiter().limitPeriodCount());
}, requestLimitDTO.getLimiter().period());
});
}
@Override
public RequestLimitType getType() {
return RequestLimitType.LEAKY_BUCKET;
}
}
令牌算法
圖解
介紹
此算法也是對于漏桶的算法的改進，這個邏輯是桶里面有一個閾值，按照一定的速率進行在桶里面存放令牌，直到令牌滿了，就不在新增令牌，然后請求每次來就去桶中獲取令牌，獲取到了，就進行處理，沒有令牌則執行拒絕策略
這個算法其實原理類似于生產者，消費者的模型，生產者按照一定的速度生成令牌，消費者可以消費數據，相對來說，這個是比較好用的
代碼
@Slf4j
@Service
public class TokenBucketRateLimitServiceImpl implements RequestLimitService {
@Autowired
private ResourcePatternResolver resourcePatternResolver;
@Autowired
private RedisService redisService;
@Resource(name = Constants.THREAD_POOL_TASK_BEAN_NAME)
private ThreadPoolTaskScheduler executor;
@Value("${limit.scan.package}")
private String scanPackage;
@Override
public boolean checkRequestLimit(RequestLimitParam param){
Object pop = redisService.listRightPop(LimitKeyConstant.QPS_TOKEN + param.getKey());
RequestLimiter limiter = param.getLimiter();
log.info("TokenBucketRateLimitServiceImpl limit period {} ms create {} total token，max token num is：{}", limiter.period(),
limiter.limitPeriodCount(), limiter.limitCount());
if (pop == null) {
log.info("TokenBucketRateLimitServiceImpl limit is empty key:{}", param.getKey());
return true;
} else {
return false;
}
}
@PostConstruct
public void init(){
// 掃描出所有使用了自定義注解并且限流類型為令牌算法的方法信息
List<RequestLimitParam> list = this.getTokenLimitList(resourcePatternResolver, RequestLimitType.TOKEN, scanPackage);
if (list.isEmpty()) {
log.info("TokenBucketRateLimitServiceImpl annotation is empty,end current task pool");
return;
}
// 每個接口方法更具注解配置信息提交定時任務，生成令牌進令牌桶
list.forEach(limit -> executor.scheduleAtFixedRate(() -> {
String key = LimitKeyConstant.QPS_TOKEN + limit.getKey();
Long tokenSize = redisService.listSize(key);
int size = tokenSize == null ? 0 : tokenSize.intValue();
if (size >= limit.getLimiter().limitCount()) {
return;
}
// 判斷添加令牌數量
int addSize = Math.min(limit.getLimiter().limitPeriodCount(), limit.getLimiter().limitCount() - size);
List<String> addList = new ArrayList<>(addSize);
for (int index = 0; index < addSize; index++) {
addList.add(UUID.randomUUID().toString());
}
redisService.listLeftPushAll(key, addList);
}, limit.getLimiter().period()));
}
@Override
public RequestLimitType getType(){
return RequestLimitType.TOKEN;
}
}
三，總結
其實這幾種算法，不能說哪一個是最好的，只能說是要的業務邏輯是什么樣的，選擇合適的限流算法來滿足自己的業務實現，沒有最優，只有最合適。
責任編輯：武曉燕
來源：
今日頭條
