Sentinel基于滑动窗口的流量统计【源码笔记】
一、问题思考
1.StatisticSlot主要职责是流量统计（为后面插槽链的流控和降级做准备），这些流量是如何统计出来的？ 2.统计出来的流量在流控中是如何使用的？
二、StatisticSlot请求流量统计
以StatisticSlot为切入点，追踪请求流量统计调用链。
1.入口
// 代码坐标：StatisticSlot#entry
public void entry(){
// 触发向下插槽执行
fireEntry(context, resourceWrapper, node, count, prioritized, args);
// 请求通过递增线程数量
node.increaseThreadNum();
// 请求通过递增请求数量
node.addPassRequest(count);
...
}
两个维度进行统计：一个是统计线程数通过StatisticNode#curThreadNum递增来完成，curThreadNum为AtomicInteger类型；另外一个是递增请求数量。
2.请求流量追踪
// 代码坐标：StatisticNode#addPassRequest
public void addPassRequest(int count) {
rollingCounterInSecond.addPass(count);
rollingCounterInMinute.addPass(count);
}
其中有两个成员变量rollingCounterInSecond和rollingCounterInMinute负责对请求流量进行统计。
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
IntervalProperty.INTERVAL);
ArrayMetric接受两个参数：int sampleCount采样窗口数量默认为2；int intervalInMs统计区间默认为1秒；rollingCounterInSecond表示近1秒的请求流量统计。
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
rollingCounterInMinute表示近1分钟的请求流量统计；ArrayMetric入参分别为60个采样窗口数量和1分钟统计区间。
// 代码坐标：ArrayMetric#addPass
public void addPass(int count) {
WindowWrap<MetricBucket> wrap = data.currentWindow();
wrap.value().addPass(count);
}
在这段代码中通过滑动窗口进行请求流量统计，详见下文分析。
三、滑动窗口流量统计
1.滑动窗口示意图
基于滑动窗口的限流，由于开始时间是浮动的，高峰流量不会出现在固定周期的开始时间段，使得整体负载趋于均衡。
时间窗口大小和统计区间可以自定义，以默认进行分析。如图所示： 统计区间：intervalInMs为1秒 滑动时间窗口大小：windowLengthInMs为500毫秒 采样窗口数量：默认2个 sampleCount=intervalInMs/windowLengthInMs=2 采样数据数组：数据大小默认2 使用数组来封装（array）数组大小与采样窗口数量相同 采样数据数组下标：
idx=(int)(timeId % array.length())
long timeId = timeMillis / windowLengthInMs
小结：随着时间（time）的向前推进，采样数据下标idx也在不断切换（由于2个窗口在0和1之间切换）；根据下标进而获取采样数据，通过比较当前时间与采样数据中的窗口开始时间，确定当前时间是否属于该滑动窗口以及该采样数据的窗口是否过期；通过不断重置与更新采样数据的值实现统计数据的动态变化。
2.代码分析
通过分析代码详细解释滑动窗口示意图。
接上面代码
// 代码坐标：ArrayMetric#addPass
public void addPass(int count) {
WindowWrap<MetricBucket> wrap = data.currentWindow();
wrap.value().addPass(count);
}
注：addPass通过CAS对数据进行递增UNSAFE.compareAndSwapLong，详见：LongAdder#add(long x)。
根据给定的时间戳获取对应的滑动窗口数据。
// 代码坐标：LeapArray#currentWindow
public WindowWrap<T> currentWindow(long timeMillis) {
// 计算数组array对应的下标
int idx = calculateTimeIdx(timeMillis);
// 计算窗口开始时间（剔除余数）
long windowStart = calculateWindowStart(timeMillis);
// 一直运行直到返回窗口
while (true) {
// 获取数据对应的采样数据
WindowWrap<T> old = array.get(idx);
// 没有窗口统计数据
if (old == null) {
// 构造窗口
WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
return window;
// 处于同一个时间窗口
} else if (windowStart == old.windowStart()) {
return old;
// 时间已经推进到下一个窗口原来窗口过期
} else if (windowStart > old.windowStart()) {
// 重置时间窗口
return resetWindowTo(old, windowStart);
}
}
}
3.获取滑动窗口流程图
4.数据打印测试
测试代码
@Test
public void testGetWindow() throws InterruptedException {
while(true){
BucketLeapArray leapArray = new BucketLeapArray(sampleCount, intervalInMs);
long time = TimeUtil.currentTimeMillis();
WindowWrap<MetricBucket> window = leapArray.currentWindow(time);
System.out.println("输入时间time:"+time);
System.out.println("统计区间intervalInMs:"+intervalInMs);
System.out.println("滑动窗口长度windowLengthInMs:" + window.windowLength());
System.out.println("采样窗口数量sampleCount:" + sampleCount);
System.out.println("当前时间对应窗口的开始时间windowStart:"+window.windowStart());
System.out.println("---------------------------------------");
Thread.sleep(new Random().nextInt(2000) );
}
}
结果输出：
采样数据下标idx：0
输入时间time:1570325921114
统计区间intervalInMs:1000
滑动窗口长度windowLengthInMs:500
采样窗口数量sampleCount:2
当前时间对应窗口的开始时间windowStart:1570325921000
---------------------------------------
采样数据下标idx：1
输入时间time:1570325921916
统计区间intervalInMs:1000
滑动窗口长度windowLengthInMs:500
采样窗口数量sampleCount:2
当前时间对应窗口的开始时间windowStart:1570325921500
---------------------------------------
采样数据下标idx：1
输入时间time:1570325923596
统计区间intervalInMs:1000
滑动窗口长度windowLengthInMs:500
采样窗口数量sampleCount:2
当前时间对应窗口的开始时间windowStart:1570325923500
---------------------------------------
采样数据下标idx：0
输入时间time:1570325925041
统计区间intervalInMs:1000
滑动窗口长度windowLengthInMs:500
采样窗口数量sampleCount:2
当前时间对应窗口的开始时间windowStart:1570325925000
四、滑动窗口流量数据使用
FlowSlot职责在于比较流控规则与已统计的流量，未达到阀值则放行；达到阀值则触发流控，以此为例跟踪下如何使用滑动窗口统计的流量。
1.代码调用链
1.FlowSlot#entry
2.FlowSlot#checkFlow
3.FlowRuleChecker#checkFlow
4.FlowRuleChecker#passLocalCheck
5.TrafficShapingController#canPass
6.DefaultController#canPass
2.代码分析
// 流控判断
public boolean canPass(Node node, int acquireCount, boolean prioritized){
// 获取已使用的tokens（线程数或者QPS）
int curCount = avgUsedTokens(node);
// acquireCount入参为1
// 超过阀值触发流控
if (curCount + acquireCount > count) {
...
return false;
}
return true;
}
流控判断代码中通过avgUsedTokens获取当前流量的平均值curCount，然后与阀值count进行比较。
// 以请求流量为例跟踪
private int avgUsedTokens(Node node) {
if (node == null) {
return DEFAULT_AVG_USED_TOKENS;
}
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}
public double passQps() {
return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}
rollingCounterInSecond.getWindowIntervalInSec()为滑动窗口统计区间默认为1秒，rollingCounterInSecond.pass()代码如下：
public long pass() {
data.currentWindow();
long pass = 0;
// 获取所有有效的滑动窗口
List<MetricBucket> list = data.values();
for (MetricBucket window : list) {
pass += window.pass();
}
return pass;
}
小结：ArrayMetric#pass通过获取当前所有有效的滑动窗口，计算每个窗口统计的流量（window.pass）之和即为该统计区间的总流量。统计区间的总流量（默认2个滑动窗口流量之和）除以统计区间时间（1秒）即为该统计区间的平均流量。