Sentinel滑动时间窗源码 算法原理 核心源码

发表于

算法原理

1、（固定）时间窗限流算法

  • 特点：系统自动选定一个时间窗口的起始零点，然后按照固定长度将时间轴划分为若干定长的时间窗口。
  • 原理：判断请求到达的时间点所在的时间窗口当前统计的数据是否超出阈值。
  • 缺点：跨窗口范围的数据超出阈值问题。
    1.png

2、 滑动时间窗限流算法

  • 特点：没有划分固定的时间窗起点与终点，而是将每一次请求的到来时间点作为统计时间窗的终点，起点则是终点向前推时间窗长度的时间点。
  • 原理：判断 请求到来时间点 - 窗口长度 范围内数据是否超出阈值。
  • 缺点：重复统计、性能问题。
    1.png

3、（改进）滑动时间窗限流算法

  • 特点：固定时间窗+滑动时间窗结合，时间窗分为若干“样本窗口”。
  • 原理：每个样本窗口会统计一次数据并记录下来。当一个请求到达时，会统计出当前请求时间点所在样本窗口中的流量数据，然后加上时间窗中其它样本窗口的统计数据，判断是否超出阈值。
  • 缺点：不够精确。只是时间窗口被细粒度化了，不准确性降低很多而已。
    1.png

核心源码

sentinel滑动时间窗.png

一、数据统计

核心类：

  • StatisticSlot - 统计入口
  • DefaultNode - 实际入口
  • StatisticNode - 统计节点
  • ArrayMetric - 使用数组保存数据的计量器类
  • LeapArray - 样本窗口数组（环性数组）
  • BucketLeapArray - 重置样本窗口
  • WindowWrap - 样本窗口（泛型T为MetricBucket）
  • MetricBucket - 统计数据封装类（多维度，维度类型在MetricEvent枚举）

1、StatisticSlot - 统计入口

用于记录、统计不同纬度的 runtime 指标监控信息；（做实时统计）

  • 线程数：内部维护一个LongAdder来进行当前线程数的统计，每进一个请求+1，每释放一个请求-1。
  • QPS：通过滑动时间窗统计请求数量是否超过阈值。

主要做3件事：

  • 1、通过node中的当前的实时统计指标信息进行规则校验
  • 2、如果通过了校验，则重新更新node中的实时指标数据
  • 3、如果被block或出现了异常了，则重新更新node中block的指标或异常指标
1
2
3
4
5
6
7
8
9
10
11
12
java复制代码@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args) throws Throwable {
    try {
        // 向后传递：调用SlotChain中后续的所有Slot，完成所有规则检测（执行过程中可能回抛出异常：如，BlockException）
        fireEntry(context, resourceWrapper, node, count, prioritized, args);

        // 前面所有规则检测通过：对DefaultNode添加线程数和qps（通过的请求数量：滑动时间窗）
        node.increaseThreadNum();
        node.addPassRequest(count);
        ... ...
}

2、DefaultNode - 实际入口

统计资源当前入口和全局数据

  • DefaultNode：保存着某个resource在某个context中的实时指标，每个DefaultNode都指向一个ClusterNode
  • ClusterNode：保存着某个resource在所有的context中实时指标的总和，同样的resource会共享同一个ClusterNode，不管他在哪个context中
1
2
3
4
5
6
7
java复制代码@Override
public void addPassRequest(int count) {
    // 增加当前入口defaultNode统计数据(调用父类StatisticNode)
    super.addPassRequest(count);
    // 增加当前资源的clusterNode的全局统计数据(背后也是调用父类StatisticNode)
    this.clusterNode.addPassRequest(count);
}

3、StatisticNode - 统计节点

滑动计数器按 秒/分 分别增加统计数据

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码// 定义一个使用数组保存数据的计量器：样本窗口数-2、时间窗默认值-1000ms
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(
    SampleCountProperty.SAMPLE_COUNT,
    IntervalProperty.INTERVAL
);
        
@Override
public void addPassRequest(int count) {
    // 滑动计数器：秒/分 增加统计数据
    rollingCounterInSecond.addPass(count);
    rollingCounterInMinute.addPass(count);
}

4、ArrayMetric - 使用数组保存数据的计量器类

按秒/分统计数据并记录到当前样本窗口

1
2
3
4
5
6
7
java复制代码@Override
public void addPass(int count) {
    // 获取当前时间点所在的样本窗口
    WindowWrap<MetricBucket> wrap = data.currentWindow();
    // 最终增加qps值位置：将当前请求计数量添加到当前样本窗口统计数据中
    wrap.value().addPass(count);
}

5、LeapArray - 样本窗口数组（环性数组）

获取当前时间点所在的样本窗口（LeapArray采用了一个环性数组的数据结构，和一致性hash算法的图类似）

image.png

  • 1.根据当前时间，算出该时间的timeId，并根据timeId算出当前窗口在采样窗口数组中的索引idx（时间每增长一个windowLength的长度，timeId就加1：但是idx不会增长，只会在0和1之间变换，因为array数组的长度是2）
  • 2.根据当前时间算出当前窗口的应该对应的开始时间time，以毫秒为单位
  • 3.根据索引idx，在采样窗口数组中取得一个时间窗口old，然后判断处理
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
csharp复制代码public WindowWrap<T> currentWindow(long timeMillis) {
    if (timeMillis < 0) {
        return null;
    }

    // 1、计算：当前时间所在样本窗口id，即在计算数组leapArray中的索引(  (timeMillis / windowLengthInMs) % array.length() )
    int idx = calculateTimeIdx(timeMillis);
    // 2、计算：当前样本窗口开始时间点（timeMillis - timeMillis % windowLengthInMs）
    long windowStart = calculateWindowStart(timeMillis);
    while (true) {
        // 3、获取当前时间所在的样本窗口
        WindowWrap<T> old = array.get(idx);
        if (old == null) {
            // 创建时间窗
            WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
            if (array.compareAndSet(idx, null, window)) { // cas方式
                // Successfully updated, return the created bucket.
                return window;
            } else {
                // Contention failed, the thread will yield its time slice to wait for bucket available.
                Thread.yield();
            }
        } else if (windowStart == old.windowStart()) {
            // 当前样本窗口起始时间=计算出的：说明是同一个样本窗口
            return old;
        } else if (windowStart > old.windowStart()) {
            // 计算出的样本窗口已经过时（环形：已经下一圈）：重置原时间窗口（替换老的样本窗口）
            if (updateLock.tryLock()) {
                try {
                    return resetWindowTo(old, windowStart);
                } finally {
                    updateLock.unlock();
                }
            } else {
                Thread.yield();
            }
        } else if (windowStart < old.windowStart()) {
            // 一般不会出现（时间不会倒流）：除非人为修改系统时钟
            return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
        }
    }
}

wecom-temp-ef3673adc7004939310d389dbb99d72b.png

假设：时间从0开始

  • 500ms内：时间窗口不会向前滑动（timeId1），当前窗口的开始时间（0），时间窗=timeId1+timeId2
  • 500～1000ms，时间窗口就会向前滑动到下一个（timeId2），这时会更新当前窗口的开始时间（500），时间窗=timeId1+timeId2
  • 超过1000ms时：再次进入下一个时间窗口（timeId3），更新当前窗口的开始时间（1000），时间窗（此时arrays数组中的窗口将会有一个失效）=timeId2+timeId3
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
ini复制代码/**
 * 测试代码
 */
public class Test {
    public static void main(String[] args) throws InterruptedException {
        int windowLength = 500;
        int arrayLength = 2;
        calculate(windowLength, arrayLength);
        for (int i = 0; i < 3; i++) {
            Thread.sleep(100);
            calculate(windowLength, arrayLength);
        }
        for (int i = 0; i < 3; i++) {
            Thread.sleep(500);
            calculate(windowLength, arrayLength);
        }
    }

    private static void calculate(int windowLength, int arrayLength) {
        long time = System.currentTimeMillis();
        long timeId = time / windowLength;
        long currentWindowStart = time - time % windowLength;
        int idx = (int) (timeId % arrayLength);
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        System.out.println("time=" + formatter.format(time) + ",currentWindowStart=" + currentWindowStart + ",timeId=" + timeId + ",idx=" + idx);
    }
}

结果：
time=2021-07-15 20:06:17.822,currentWindowStart=1626350777500,timeId=3252701555,idx=1
time=2021-07-15 20:06:17.954,currentWindowStart=1626350777500,timeId=3252701555,idx=1
time=2021-07-15 20:06:18.054,currentWindowStart=1626350778000,timeId=3252701556,idx=0
time=2021-07-15 20:06:18.157,currentWindowStart=1626350778000,timeId=3252701556,idx=0
time=2021-07-15 20:06:18.658,currentWindowStart=1626350778500,timeId=3252701557,idx=1
time=2021-07-15 20:06:19.159,currentWindowStart=1626350779000,timeId=3252701558,idx=0
time=2021-07-15 20:06:19.662,currentWindowStart=1626350779500,timeId=3252701559,idx=1

6、BucketLeapArray - 重置样本窗口

计算出的样本窗口已经过时：重置原时间窗口（替换老的样本窗口）

1
2
3
4
5
6
7
8
java复制代码@Override
protected WindowWrap<MetricBucket> resetWindowTo(WindowWrap<MetricBucket> w, long startTime) {
    // 更新窗口起始时间：仅把数据替换
    w.resetTo(startTime);
    // 将每个维度统计数据清零
    w.value().reset();
    return w;
}

7、MetricBucket - 统计数据封装类

pass维度增加

1
2
3
4
5
6
7
8
csharp复制代码public void addPass(int n) {
    // pass维度增加
    add(MetricEvent.PASS, n);
}
public MetricBucket add(MetricEvent event, long n) {
    counters[event.ordinal()].add(n);
    return this;
}

二、数据使用（qps例子）

核心类：

  • DefaultController - 入口
  • StatisticNode - 实际入口
  • ArrayMetric - 使用数组保存数据的计量器类
  • LeapArray - 样本窗口数组（环性数组）

1、DefaultController - 入口

获取当前时间窗已统计数据

1
2
3
4
5
6
7
csharp复制代码private int avgUsedTokens(Node node) {
    if (node == null) {
        // 未做统计工作：返回0
        return DEFAULT_AVG_USED_TOKENS;
    }
    return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}

2、StatisticNode - 实际入口

获取通过qps数量

1
2
3
4
5
6
7
8
9
10
11
java复制代码// 定义一个使用数组保存数据的计量器：样本窗口数-2、时间窗默认值-1000ms
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(
    SampleCountProperty.SAMPLE_COUNT,
    IntervalProperty.INTERVAL
);
    
@Override
public double passQps() {
    // 时间窗场景：当前时间窗中统计的通过的请求数量/时间窗长度（秒）
    return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}

3、ArrayMetric - 使用数组保存数据的计量器类

汇总pass数据：所有样本窗口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
typescript复制代码@Override
public long pass() {
    // 更新array中当前时间点所在的样本窗口实例中的数据
    data.currentWindow();
    long pass = 0;
    // 获取：当前时间窗口中的所有样本窗口统计的value，记录到result中
    List<MetricBucket> list = data.values();

    // 汇总pass数据
    for (MetricBucket window : list) {
        pass += window.pass();
    }
    return pass;
}

4、LeapArray - 样本窗口数组（环性数组）

汇总样本窗口实例：要过滤过时的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
arduino复制代码public List<T> values(long timeMillis) {
    if (timeMillis < 0) {
        return new ArrayList<T>();
    }
    int size = array.length();
    List<T> result = new ArrayList<T>(size);

    // 遍历array中每个样本窗口实例，并汇总result
    for (int i = 0; i < size; i++) {
        WindowWrap<T> windowWrap = array.get(i);
        // 若当前遍历实例：空/过时
        if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
            continue;
        }
        result.add(windowWrap.value());
    }
    return result;
}

public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {
    // 当前时间与当前样本窗口时间差 > 窗口时间 ： 说明过时（环形：已经下一圈）
    return time - windowWrap.windowStart() > intervalInMs;
}

参考资料

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%