其實(shí)這幾種算法，不能說(shuō)哪一個(gè)是最好的，只能說(shuō)是要的業(yè)務(wù)邏輯是什么樣的，選擇合適的限流算法來(lái)滿足自己的業(yè)務(wù)實(shí)現(xiàn)，沒(méi)有最優(yōu)，只有最合適。
一、概述
限流其實(shí)就是對(duì)服務(wù)的請(qǐng)求做一下QPS的控制，對(duì)于有些免登錄的接口需要做一下訪問(wèn)的限制，不能無(wú)限制的去請(qǐng)求接口，不然的話會(huì)給服務(wù)器造成很大的壓力，而且我們也希望一些接口做一下控制，控制請(qǐng)求量，這樣我們就可以做一個(gè)plugin對(duì)服務(wù)做限流操作，超出限流就返回請(qǐng)求失敗，保證系統(tǒng)的穩(wěn)定運(yùn)行。主要概念就是閾值以及拒絕策略，實(shí)際中需要用到限流的的比如，驗(yàn)證碼，白名單，當(dāng)然也有容器的限流，比如nginx就是比較常用的，可以做一下簡(jiǎn)單的處理。
二、限流算法類(lèi)型
幾種算法的使用，一些基礎(chǔ)代碼如下
限流代碼基礎(chǔ)類(lèi)
@RequestLimiter
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Order(Ordered.HIGHEST_PRECEDENCE)
public @interface RequestLimiter {
/**
* 限流類(lèi)型 ，具體見(jiàn)枚舉類(lèi) RequestLimitType
*/
RequestLimitType type() default RequestLimitType.TOKEN;
/**
* 限流訪問(wèn)數(shù)
*/
int limitCount() default 100;
/**
* 限流時(shí)間段
*/
long time() default 60;
/**
* 限流時(shí)間段 時(shí)間單位
*/
TimeUnit unit() default TimeUnit.SECONDS;
/**
* 漏出或者生成令牌時(shí)間間隔，單位 毫秒 (當(dāng)type為T(mén)OKEN、LEAKY_BUCKET時(shí)生效)
*/
long period() default 1000;
/**
* 每次生成令牌數(shù)或者漏出水滴數(shù) (當(dāng)type為T(mén)OKEN、LEAKY_BUCKET時(shí)生效)
*/
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:";
/**
* 滑動(dòng)窗口鍵名
*/
public static final String QPS_SLIDE_WINDOW = "request:limit:qps:slideWindow:";
}RequestLimitType
public enum RequestLimitType {
/**
* 令牌算法
*/
TOKEN(1, "令牌算法"),
/**
* 漏桶算法
*/
LEAKY_BUCKET(2, "漏桶算法"),
/**
* 固定窗口
*/
FIXED_WINDOW(3, "固定窗口"),
/**
* 滑動(dòng)窗口
*/
SLIDE_WINDOW(4, "滑動(dòng)窗口");
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;
/**
* 切入點(diǎn)
*/
@Pointcut(value = "@annotation(com.common.limit.annotation.RequestLimiter)")
public void requestLimit(){
// 切入點(diǎn)方法
}
/**
* 前置切點(diǎn)
*
* @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("請(qǐng)求過(guò)于頻繁，請(qǐng)稍后再重試！");
}
}
}
}
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);
}
}
/**
* 構(gòu)建service
*
* @param type 限流類(lèi)型
* @return
public RequestLimitService build(RequestLimitType type){
return MAP.get(type);
}
}
RequestLimitService
public interface RequestLimitService {
/**
* 檢測(cè)是否限流
*
* @param param 限流參數(shù)
* @return
boolean checkRequestLimit(RequestLimitParam param);
/**
* 獲取當(dāng)前限流類(lèi)型
*
* @return
RequestLimitType getType();
/**
* 獲取帶注解方法列表
*
* @param resourcePatternResolver 資源查詢
* @param limitType 注解類(lèi)型
* @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();
}
}
}固定時(shí)間窗口算法
圖解
介紹
其實(shí)就是原子計(jì)數(shù)法，就是在固定時(shí)間內(nèi)，允許請(qǐng)求量是多少，每次請(qǐng)求就在計(jì)數(shù)器上加1，設(shè)置計(jì)數(shù)器的過(guò)期時(shí)間，當(dāng)計(jì)數(shù)器的閾值達(dá)到限流配置的數(shù)時(shí)候，就執(zhí)行拒絕策略，超過(guò)了時(shí)間，計(jì)數(shù)器就會(huì)重新歸0。
比如上圖中，會(huì)限制在每秒限制請(qǐng)求數(shù)為2，就是在每秒的時(shí)間會(huì)限制請(qǐng)求為2，但是會(huì)出現(xiàn)極端的情況，比如在前一個(gè)時(shí)間段中的前500ms和后500ms，請(qǐng)求數(shù)都是2，這樣就會(huì)看到在這一秒內(nèi)是有4個(gè)請(qǐng)求的，這就是會(huì)出現(xiàn)請(qǐng)求的問(wèn)題，當(dāng)然這也是最簡(jiǎn)單的限流算法。
代碼
@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());
// 檢測(cè)是否到達(dá)限流值
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;
}
}
滑動(dòng)時(shí)間窗口算法
圖解
介紹
滑動(dòng)時(shí)間窗口算法，其實(shí)就是對(duì)固定窗口的改進(jìn)，知道了固定時(shí)間窗口會(huì)出現(xiàn)極端的情況，那滑動(dòng)就在下一個(gè)臨界的時(shí)候，進(jìn)行處理時(shí)間，其實(shí)就是在某一段時(shí)間進(jìn)行處理時(shí)間。
比如上圖中每 500ms 滑動(dòng)一次窗口，可以發(fā)現(xiàn)窗口滑動(dòng)的間隔越短，時(shí)間窗口的臨界突變問(wèn)題發(fā)生的概率也就越小，不過(guò)只要有時(shí)間窗口的存在，還是有可能發(fā)生時(shí)間窗口的臨界突變問(wèn)題。
這個(gè)是記錄下所有的請(qǐng)求時(shí)間點(diǎn)，新請(qǐng)求先判斷最近指定時(shí)間范圍內(nèi)的請(qǐng)求數(shù)量是否超過(guò)指定閾值，來(lái)確定是否達(dá)到限流，雖然沒(méi)有時(shí)間窗口突變的問(wèn)題，限流比較準(zhǔn)確，但是要記錄下每次請(qǐng)求的時(shí)間點(diǎn)，所以占用的內(nèi)存較多。
代碼
@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);
// 清除有效期外的數(shù)據(jù)
redisService.setRemoveRangeByScore(key, 0, current - duringTime - 1f);
log.info("SlideWindowRateLimitServiceImpl time：{} unit：{} allow visit {}", limiter.time(), limiter.unit(), limiter.limitCount());
// 檢測(cè)是否到達(dá)限流值
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;
}
}
漏桶算法
圖解
介紹
此算法就是定義一個(gè)桶的容量，然后每次的請(qǐng)求過(guò)來(lái)都放在桶里面，一直等到桶滿了以后就會(huì)執(zhí)行拒絕策略，然后在桶不滿的情況下，會(huì)按照固定的速率去執(zhí)行請(qǐng)求，其實(shí)就是按照固定流速去執(zhí)行請(qǐng)求，保證單位時(shí)間內(nèi)的執(zhí)行請(qǐng)求量是固定的。
漏桶就是按照某一個(gè)請(qǐng)求的穩(wěn)定的速度處理發(fā)來(lái)的請(qǐng)求數(shù)量，可以很好地保證系統(tǒng)的穩(wěn)定運(yùn)行，只能平穩(wěn)處理請(qǐng)求，這也是他的一個(gè)缺點(diǎn)，不能處理面對(duì)突然來(lái)的高的請(qǐng)求量，會(huì)導(dǎo)致請(qǐng)求一直處于哎隊(duì)列等待中，不能面對(duì)高并發(fā)下的請(qǐng)求處理，比較保守的處理邏輯
代碼
@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;
}
}
/**
* 定數(shù)流出令牌
*/
@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;
}
}
令牌算法
圖解
介紹
此算法也是對(duì)于漏桶的算法的改進(jìn)，這個(gè)邏輯是桶里面有一個(gè)閾值，按照一定的速率進(jìn)行在桶里面存放令牌，直到令牌滿了，就不在新增令牌，然后請(qǐng)求每次來(lái)就去桶中獲取令牌，獲取到了，就進(jìn)行處理，沒(méi)有令牌則執(zhí)行拒絕策略
這個(gè)算法其實(shí)原理類(lèi)似于生產(chǎn)者，消費(fèi)者的模型，生產(chǎn)者按照一定的速度生成令牌，消費(fèi)者可以消費(fèi)數(shù)據(jù)，相對(duì)來(lái)說(shuō)，這個(gè)是比較好用的
代碼
@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(){
// 掃描出所有使用了自定義注解并且限流類(lèi)型為令牌算法的方法信息
List<RequestLimitParam> list = this.getTokenLimitList(resourcePatternResolver, RequestLimitType.TOKEN, scanPackage);
if (list.isEmpty()) {
log.info("TokenBucketRateLimitServiceImpl annotation is empty,end current task pool");
return;
}
// 每個(gè)接口方法更具注解配置信息提交定時(shí)任務(wù)，生成令牌進(jìn)令牌桶
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;
}
// 判斷添加令牌數(shù)量
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;
}
}
三，總結(jié)
其實(shí)這幾種算法，不能說(shuō)哪一個(gè)是最好的，只能說(shuō)是要的業(yè)務(wù)邏輯是什么樣的，選擇合適的限流算法來(lái)滿足自己的業(yè)務(wù)實(shí)現(xiàn)，沒(méi)有最優(yōu)，只有最合適。
