本文的思路是利用AOP技术+自定义注解实现对特定的方法或接口进行限流。目前通过查阅相关资料,整理出三种类型限流方法,分别为基于guava限流实现、基于sentinel限流实现、基于Semaphore的实现。
一、限流常用的算法
1.1令牌桶算法
令牌桶算法是目前应用最为广泛的限流算法,顾名思义,它有以下两个关键角色。
令牌 :获取到令牌的Request才会被处理,其他Requests要么排队要么被直接丢弃;
桶 :用来装令牌的地方,所有Request都从这个桶里面获取令牌。
令牌桶主要涉及到2个过程,即令牌的生成,令牌的获取。
1.2漏桶算法
漏桶算法的前半段和令牌桶类似,但是操作的对象不同,结合下图进行理解。令牌桶是将令牌放入桶里,而漏桶是将访问请求的数据包放到桶里。同样的是,如果桶满了,那么后面新来的数据包将被丢弃。
1.3滑动时间窗口
根据下图,简单描述下滑动时间窗口这种过程:
黑色大框为时间窗口,可以设定窗口时间单位为5秒,它会随着时间推移向后滑动。我们将窗口内的时间划分为五个小格子,每个格子代表1秒钟,同时这个格子还包含一个计数器,用来计算在当前时间内访问的请求数量。那么这个时间窗口内的总访问量就是所有格子计数器累加后的数值;
比如说,我们在每一秒内有5个用户访问,第5秒内有10个用户访问,那么在0到5秒这个时间窗口内访问量就是15。如果我们的接口设置了时间窗口内访问上限是20,那么当时间到第六秒的时候,这个时间窗口内的计数总和就变成了10,因为1秒的格子已经退出了时间窗口,因此在第六秒内可以接收的访问量就是20-10=10个。
滑动窗口其实也是一种计算器算法,它有一个显著特点,当时间窗口的跨度越长时,限流效果就越平滑。打个比方,如果当前时间窗口只有两秒,而访问请求全部集中在第一秒的时候,当时间向后滑动一秒后,当前窗口的计数量将发生较大的变化,拉长时间窗口可以降低这种情况的发生概率。
二、常用限流方案的实现
2.1基于guava限流实现
guava为谷歌开源的一个比较实用的组件,利用这个组件可以帮助开发人员完成常规的限流操作。其限流工具类——RateLimiter,通过使用该工具类来实现限流功能,RateLimiter是基于“令牌通算法”来实现限流的。
2.1.1引入guava依赖
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
2.1.2TokenBucketLimiter自定义限流注解
/**
* 基于guava的令牌桶
*/
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface TokenBucketLimiter {
int value() default 50;
}
2.1.3GuavaLimiterAop切面
@Aspect
@Component
@Slf4j
public class GuavaLimiterAop {
private final Map<String, RateLimiter> rateLimiters = new ConcurrentHashMap<String, RateLimiter>();
@Pointcut("@annotation(com.mylimit.annotation.TokenBucketLimiter)")
public void aspect() {
}
@Around(value = "aspect()")
public Object around(ProceedingJoinPoint point) throws Throwable {
log.debug("准备限流");
Object target = point.getTarget();
String targetName = target.getClass().getName();
String methodName = point.getSignature().getName();
Object[] arguments = point.getArgs();
Class<?> targetClass = Class.forName(targetName);
Class<?>[] argTypes = ReflectUtils.getClasses(arguments);
Method method = targetClass.getDeclaredMethod(methodName, argTypes);
// 获取目标method上的限流注解@Limiter
TokenBucketLimiter limiter = method.getAnnotation(TokenBucketLimiter.class);
RateLimiter rateLimiter;
Object result = null;
if (null != limiter) {
// 以 class + method + parameters为key,避免重载、重写带来的混乱
String key = targetName + "." + methodName + Arrays.toString(argTypes);
rateLimiter = rateLimiters.get(key);
if (null == rateLimiter) {
// 获取限定的流量
// 为了防止并发
rateLimiters.putIfAbsent(key, RateLimiter.create(limiter.value()));
rateLimiter = rateLimiters.get(key);
}
boolean b = rateLimiter.tryAcquire();
if (b) {
log.debug("得到令牌,准备执行业务");
result = point.proceed();
} else {
HttpServletResponse resp = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();
JSONObject jsonObject = new JSONObject();
jsonObject.put("success", false);
jsonObject.put("msg", "限流中");
try {
output(resp, jsonObject.toJSONString());
} catch (Exception e) {
log.error("error,e:{}", e);
}
}
} else {
result = point.proceed();
}
log.debug("退出限流");
return result;
}
public void output(HttpServletResponse response, String msg) throws IOException {
response.setContentType("application/json;charset=UTF-8");
ServletOutputStream outputStream = null;
try {
outputStream = response.getOutputStream();
outputStream.write(msg.getBytes("UTF-8"));
} catch (IOException e) {
e.printStackTrace();
} finally {
outputStream.flush();
outputStream.close();
}
}
2.2基于sentinel限流实现
以前只知道sentinel是需要结合springcloud-alibaba框架一起使用,而且与框架集成之后,可以配合控制台一起使用达到更好的效果。实际上,sentinel官方也提供了相对原生的SDK可供使用。
2.2.1引入sentinel核心依赖包
<dependency>
<groupId>com.alibaba.csp</groupId>
<artifactId>sentinel-core</artifactId>
<version>1.8.0</version>
</dependency>
2.2.2SentinelLimiter自定义限流注解
/**
* 基于sentinel限流
*/
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface SentinelLimiter {
String resourceName();
int value() default 50;
}
2.2.3SentinelLimiterAop自定义AOP类
@Aspect
@Component
public class SentinelLimiterAop {
private static void initFlowRule(String resourceName, int limitCount) {
List<FlowRule> rules = new ArrayList<>();
FlowRule rule = new FlowRule();
//设置受保护的资源
rule.setResource(resourceName);
//设置流控规则 QPS
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
//设置受保护的资源阈值
rule.setCount(limitCount);
rules.add(rule);
//加载配置好的规则
FlowRuleManager.loadRules(rules);
}
@Pointcut(value = "@annotation(com.mylimit.annotation.SentinelLimiter)")
public void rateLimit() {
}
@Around("rateLimit()")
public Object around(ProceedingJoinPoint joinPoint) {
//1、获取当前的调用方法
Method currentMethod = getCurrentMethod(joinPoint);
if (Objects.isNull(currentMethod)) {
return null;
}
//2、从方法注解定义上获取限流的类型
String resourceName = currentMethod.getAnnotation(SentinelLimiter.class).resourceName();
if (StringUtils.isEmpty(resourceName)) {
throw new RuntimeException("资源名称为空");
}
int limitCount = currentMethod.getAnnotation(SentinelLimiter.class).value();
initFlowRule(resourceName, limitCount);
Entry entry = null;
Object result = null;
try {
entry = SphU.entry(resourceName);
try {
result = joinPoint.proceed();
} catch (Throwable throwable) {
throwable.printStackTrace();
}
} catch (BlockException ex) {
// 资源访问阻止,被限流或被降级
// 在此处进行相应的处理操作
System.out.println("blocked");
return "被限流了";
} catch (Exception e) {
Tracer.traceEntry(e, entry);
} finally {
if (entry != null) {
entry.exit();
}
}
return result;
}
private Method getCurrentMethod(JoinPoint joinPoint) {
Method[] methods = joinPoint.getTarget().getClass().getMethods();
Method target = null;
for (Method method : methods) {
if (method.getName().equals(joinPoint.getSignature().getName())) {
target = method;
break;
}
}
return target;
}
}
2.3基于Semaphore的aop实现
2.3.1自定义SemaphoreLimiter注解
/**
* 基于java自带的Semaphore限流
*/
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface SemaphoreLimiter {
int value() default 50;
}
2.3.2自定义SemaphoreLimiterAop限流切面
@Aspect
@Component
@Slf4j
public class SemaphoreLimiterAop {
private final Map<String, Semaphore> semaphores = new ConcurrentHashMap<String, Semaphore>();
private final static Logger LOG = LoggerFactory.getLogger(SemaphoreLimiterAop.class);
@Pointcut("@annotation(com.mylimit.annotation.SemaphoreLimiter)")
public void aspect() {
}
@Around(value = "aspect()")
public Object around(ProceedingJoinPoint point) throws Throwable {
log.debug("进入限流aop");
Object target = point.getTarget();
String targetName = target.getClass().getName();
String methodName = point.getSignature().getName();
Object[] arguments = point.getArgs();
Class<?> targetClass = Class.forName(targetName);
Class<?>[] argTypes = ReflectUtils.getClasses(arguments);
Method method = targetClass.getDeclaredMethod(methodName, argTypes);
// 获取目标method上的限流注解@Limiter
SemaphoreLimiter limiter = method.getAnnotation(SemaphoreLimiter.class);
Object result = null;
if (null != limiter) {
// 以 class + method + parameters为key,避免重载、重写带来的混乱
String key = targetName + "." + methodName + Arrays.toString(argTypes);
// 获取限定的流量
Semaphore semaphore = semaphores.get(key);
if (null == semaphore) {
semaphores.putIfAbsent(key, new Semaphore(limiter.value()));
semaphore = semaphores.get(key);
}
try {
semaphore.acquire();
result = point.proceed();
} finally {
if (null != semaphore) {
semaphore.release();
}
}
} else {
result = point.proceed();
}
log.debug("退出限流");
return result;
}
}
2.4配置spring.factories
在自定义starter的resources目录下,新创建spring.factories,
内容如下
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.mylimit.aspect.SemaphoreLimiterAop,\
com.mylimit.aspect.GuavaLimiterAop,\
com.mylimit.aspect.SemaphoreLimiterAop
三、代码测试
3.1LimitController测试
先在主starter中引入自定义的starter,这里引入自定义的starter为limit-spring-boot-starter
<dependency>
<groupId>com.mylimit</groupId>
<artifactId>limit-spring-boot-starter</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
LimitController
@RestController
@RequestMapping("/limitController")
public class LimitController {
@GetMapping("/query")
@TokenBucketLimiter(1)
public String queryUser() {
return "queryUser";
}
}
3.3测试结果验证
通过在浏览器网页或者postman测试,目前设置限流QPS为1,发现快速重复多次请求,就会触发限流。
四、项目结构及源码下载
源码下载地址,欢迎Star!!!
mystarter
参考资料
实现自定义Spring Boot Starter
springboot 通用限流方案设计与实现
Caused by: java.lang.ClassNotFoundException: io.r2dbc.spi.ValidationDepth 启动报错
使用Guava实现限流器