滑动窗口算法实现单位时间API限流_滑动窗口实现qps限制

1. 限流

限流顾名思义，就是对请求或并发数进行限制；通过对一个时间窗口内的请求量进行限制来保障系统的正常运行。如果我们的服务资源有限、处理能力有限，就需要对调用我们服务的上游请求进行限制，以防止自身服务由于资源耗尽而停止服务。

在限流中有两个概念需要了解。

• 阈值：在一个单位时间内允许的请求量。如 QPS 限制为10，说明 1 秒内最多接受 10 次请求。
• 拒绝策略：超过阈值的请求的拒绝策略，常见的拒绝策略有直接拒绝、排队等待等。

2.滑动窗口算法

滑动窗口算法是固定窗口算法的优化版本，主要是为了解决固定窗口中的零界值问题导致限流失败的问题。

优化的地方如下：

将一个时间窗口分为5份。每一份里面都有一个独立计数器c。在时间轴上的一个时间窗口内，没当请求过来的时候，就会求计数器 c1+c2+c3+c4+c5的和，当达到阀值就拒绝，没达到当前小格子里面的计数器就加1 。

过程如下：

3.代码实现

3.1 通用工具类 RateLimiterSlidingWindow

/**
     * 滑动窗口限流工具类
     * @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...
        }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

3.2 测试用例

    /**
     * 测试用例
     * @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);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

3.3 测试结果

3.4 业务实现

    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;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

3.5 测试成果

//限流控制
            AssertUtils.isTrue(rsShareDataMetadataService.controlAPI(request, dataId,appKey),GlobalErrorCodeConstants.TOO_MANY_REQUESTS,"该接口已被限流");
1
2