限流-滑动窗口计数-java实现-循环数组_滑动时间窗口限流 环形数据

一、用循环数组实现滑动窗口

1.1、实现思想

1.定义一个AtomicInteger array数组,每一个元素记录当前区间的计数

2.定义一个long数组 times,记录对应array下标元素开始的时间.

3.定义一个下标int index 记录当前正在使用的位置.

4.定义每个元素的时间区间大小span = 200 ms

index变化情况如下:

1、如果当前时间now - times[index]>span 说明当前请求计数应当位于下一个位置的元素.

index++,

如果index>=size(当前数组大小),则index=0;

2、如果now-times[index]大于传入的计数时间 如1s,则说明,该时间元素无效,重置:

times[index]=now;

array[index].set(0);

1.2、计数逻辑

1.加锁控制

2.按照上面index逻辑更新index,和对应array和times数组里面的元素信息

1.3、获取当前1s内数量

如果get方法加锁,则会影响滑动窗口性能,所以该方法不加锁,得到的值是一个近似精确的值.

实现逻辑:

1.获取到当前下标curIndex

2.设定循环次数: time = seconds(滑动窗口统计时间) * 1000 /span

3、开始循环,递减curIndex累加当前array对应的值,

判断条件

1.当前index 对应的times[index] 符合now-times[index]<seconds *1000,因为如果大于就认为超过该seconds统计区间

2、不超过time次循环

二、代码实现

链接:

数组实现

import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;


public class ArraySideWindow {
    private final AtomicInteger[] array ;
    private final long[] times;
    private final int span=200;
    private final int size ;
    private final int delta = 5;
    private final int MILL=1000;
    private final int second;
    private volatile int index;
    private ReentrantLock lock;
    public ArraySideWindow(int second){
        this.second = second;
        this.size=second* MILL / span + delta;
        this.array = new AtomicInteger[size];
        this.times = new long[size];
        this.index=0;
        for(int i=0;i<size;i++){
            this.array[i]=new AtomicInteger(0);
        }
        this.times[index]=System.currentTimeMillis();
        this.lock = new ReentrantLock(false);
    }

    public void count(){
        long now = System.currentTimeMillis();
        lock.lock();
        try{
            if(now-times[index]>span){
                index++;
                index=index<size?index:0;
            }
            if(now-times[index]>=second * MILL){
                times[index]=now;
                this.array[index].set(0);
            }
            this.array[index].incrementAndGet();
            //测试打印 忽略
            printNum(this.array);
        }catch (Exception e){

        }finally {
            lock.unlock();
        }
    }

    public int get(){
        int  curIndex = index;
        long now = System.currentTimeMillis();
        int count=0;
        int sum=0;
        while (count<5){
            if(now-times[curIndex]>=second * MILL){
                break;
            }
            sum +=array[curIndex--].get();
            if(curIndex<0){
                curIndex=size-1;
            }
            count++;
        }
        return sum;
    }

    /**
     * 测试代码
     * @param array
     */
    private synchronized void printNum(AtomicInteger[] array) {
        Random random = new Random();
        int r = random.nextInt(10);
        if(r>=3){
           return;
        }
        StringBuilder numBuilder = new StringBuilder();
        StringBuilder timeBuilder = new StringBuilder();
        for(int i=0;i<array.length;i++){
            numBuilder.append(array[i].get()).append(",");
            timeBuilder.append(times[i]).append(",");
        }
        System.out.println("-----------------------------------------");
        int preIndex = index-1<0?size-1: index-1;
        System.out.println("now:"+System.currentTimeMillis()+",curIndex:"+index+",preIndex:"+preIndex);
        System.out.println("indexTime:"+times[index]+",preIndexTime:"+times[preIndex]+",dif:"+(times[index]-times[preIndex]));

        System.out.println("Thread:"+Thread.currentThread().getName());
        System.out.println(numBuilder.toString());
        System.out.println(timeBuilder.toString());
        System.out.println("-----------------------------------------");

    }




    public static void main(String[] args) {
        ArraySideWindow window = new ArraySideWindow(1);
        AtomicInteger sum =new AtomicInteger(0);
        Random random = new Random(10);
        for(int i=0;i<30;i++){
            Thread t = new Thread(()->{
                int time = 0;
                while (time<100000){
                    int sumNow = sum.get();
                    if(sumNow>10000){
                        System.out.println("**************************************************");
                        try {
                            TimeUnit.MILLISECONDS.sleep(400+random.nextInt(30));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    window.count();
                    try {
                        TimeUnit.MILLISECONDS.sleep(2+random.nextInt(30));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Thread:"+Thread.currentThread().getName()+",num:"+window.get());
                    sum.incrementAndGet();
                }
            },"thread_"+i);
            t.start();
        }
        try {
            TimeUnit.SECONDS.sleep(100000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137