限流顾名思义，就是对请求或并发数进行限制；通过对一个时间窗口内的请求量进行限制来保障系统的正常运行。如果我们的服务资源有限、处理能力有限，就需要对调用我们服务的上游请求进行限制，以防止自身服务由于资源耗尽而停止服务。
在限流中有两个概念需要了解。
滑动窗口算法是固定窗口算法的优化版本，主要是为了解决固定窗口中的零界值问题导致限流失败的问题。
优化的地方如下：
将一个时间窗口分为5份。每一份里面都有一个独立计数器c。在时间轴上的一个时间窗口内，没当请求过来的时候，就会求计数器 c1+c2+c3+c4+c5的和，当达到阀值就拒绝，没达到当前小格子里面的计数器就加1 。
过程如下：
/**
* 滑动窗口限流工具类
* @author zyw
* @createTime: 2023/04/26 11:11
*/
@Data
public class RateLimiterSlidingWindow implements Serializable {
private static final long serialVersionUID = 920040577333729032L;
/**
* 阈值
*/
private int qps;
/**
* 时间窗口总大小（毫秒）
*/
private long windowSize;
/**
* 多少个子窗口
*/
private Integer windowCount;
/**
* 窗口列表
*/
private WindowInfo[] windowArray;
public RateLimiterSlidingWindow(int qps, long windowSize, Integer windowCount) {
this.qps = qps;
this.windowSize = windowSize;
this.windowCount = windowCount;
long currentTimeMillis = System.currentTimeMillis();
windowArray = new WindowInfo[windowCount];
for (int i = 0; i < windowArray.length; i++) {
windowArray[i] = new WindowInfo(currentTimeMillis, new AtomicInteger(0));
}
}
public RateLimiterSlidingWindow(int qps) {
this.qps = qps;
this.windowSize = 1000;
this.windowCount = 10;
windowArray = new WindowInfo[windowCount];
long currentTimeMillis = System.currentTimeMillis();
for (int i = 0; i < windowArray.length; i++) {
windowArray[i] = new WindowInfo(currentTimeMillis, new AtomicInteger(0));
}
}
/**
* 1. 计算当前时间窗口
* 2. 更新当前窗口计数 & 重置过期窗口计数
* 3. 当前 QPS 是否超过限制
*
* @return
*/
public synchronized boolean tryAcquire() {
long currentTimeMillis = System.currentTimeMillis();
// 1. 计算当前时间窗口
int currentIndex = (int)(currentTimeMillis % windowSize / (windowSize / windowCount));
// 2. 更新当前窗口计数 & 重置过期窗口计数
int sum = 0;
for (int i = 0; i < windowArray.length; i++) {
WindowInfo windowInfo = windowArray[i];
if ((currentTimeMillis - windowInfo.getTime()) > windowSize) {
windowInfo.getNumber().set(0);
windowInfo.setTime(currentTimeMillis);
}
if (currentIndex == i ) {
if (windowInfo.getNumber().get() < qps){
windowInfo.getNumber().incrementAndGet();
}else {
return false;
}
}
sum = sum + windowInfo.getNumber().get();
}
// 3. 当前 QPS 是否超过限制
return sum <= qps;
}
@Data
private class WindowInfo implements Serializable{
private static final long serialVersionUID = 940573337290329784L;
// 窗口开始时间
private Long time;
// 计数器
private AtomicInteger number;
public WindowInfo(long time, AtomicInteger number) {
this.time = time;
this.number = number;
}
// get...set...
}
}
/**
* 测试用例
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
long windowSize = 3000;
Integer windowCount = 3;
int qps = 5, count = 20, sleep = 300, success = count * sleep / 3000 * qps;
System.out.println("测试："+count*sleep+"秒，"+count+"个请求");
System.out.println("流量限制："+windowSize+"毫秒内，请求限制为"+qps+"次");
System.out.println(String.format("当前阈值限制为:%d,当前测试次数:%d,间隔:%dms,预计成功次数:%d", qps, count, sleep, success));
success = 0;
RateLimiterSlidingWindow myRateLimiter = new RateLimiterSlidingWindow(qps,windowSize,windowCount);
for (int i = 0; i < count-1; i++) {
Thread.sleep(sleep);
if (myRateLimiter.tryAcquire()) {
success++;
if (success % qps == 0) {
System.out.println(LocalTime.now() + ": success, ");
} else {
System.out.print(LocalTime.now() + ": success, ");
}
} else {
System.out.println(LocalTime.now() + ": fail");
}
}
System.out.println();
System.out.println("实际测试成功次数:" + success);
}
public static String RateLimiterSlidingWindowKey = "RateLimiterSlidingWindowKey_";
/**
* 流量控制实现业务
* @param request
* @param dataId
* @return
*/
public boolean controlAPI(HttpServletRequest request,String dataId){
//1.在Redis中查找该API滑动窗口限流工具对象是否存在
RateLimiterSlidingWindow myRateLimiter = (RateLimiterSlidingWindow)JedisUtils.getObject(RateLimiterSlidingWindowKey + dataId);
if (myRateLimiter == null){
// 同步代码块(双层检测锁)
synchronized (this) {
myRateLimiter = (RateLimiterSlidingWindow)JedisUtils.getObject(RateLimiterSlidingWindowKey + dataId);
if (myRateLimiter == null){
RsShareDataMetadata rsShareDataMetadata = get(dataId);
//如果没有配置流量控制
if ("1".equals(rsShareDataMetadata.getFlowRestriction())){
//插入我们在rsShareDataMetadata中配置好的单位限流时间以及单位时间的请求峰值
myRateLimiter = new RateLimiterSlidingWindow(rsShareDataMetadata.getAccessTimes(),rsShareDataMetadata.getAccessDays()*86400000,rsShareDataMetadata.getAccessDays());
}else {
return true;
}
}
}
}
//3.通过滑动窗口算法计算是否超过流量控制
boolean b = myRateLimiter.tryAcquire();
//4.更新Redis
JedisUtils.setObject(RateLimiterSlidingWindowKey + dataId,myRateLimiter,0);
return b;
}
//限流控制
AssertUtils.isTrue(rsShareDataMetadataService.controlAPI(request, dataId,appKey),GlobalErrorCodeConstants.TOO_MANY_REQUESTS,"该接口已被限流");
