正文
Sentinel源码分析-滑动窗口统计原理
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
滑动窗口技术是Sentinel比较关键的核心技术，主要用于数据统计
通过分析
StatisticSlot
来慢慢引出这个概念
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
try {
// Do some checking.
fireEntry(context, resourceWrapper, node, count, prioritized, args);
// Request passed, add thread count and pass count.
node.increaseThreadNum();
node.addPassRequest(count);
......
}
......
从代码中可以看出，在其他Slot通过后，会调用node进行计数，我们来看
node.addPassRequest(count);
, 由于我们使用的是FlowQpsDemo
@Override
public void addPassRequest(int count) {
super.addPassRequest(count);
this.clusterNode.addPassRequest(count);
}
直接看
this.clusterNode.addPassRequest(count);
因为clusterNode是default模式下主要用得统计数据node，而它继承于StatisticsNode,于是调用的是
//StatisticsNode.java
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
@Override
public void addPassRequest(int count) {
rollingCounterInSecond.addPass(count);
rollingCounterInMinute.addPass(count);
}
两个成员类似，我们看其中之一
rollingCounterInMinute.addPass
，最终会调用
// ArrayMetric.java
this.data = new BucketLeapArray(sampleCount, intervalInMs);
@Override
public void addPass(int count) {
//核心方法
WindowWrap<MetricBucket> wrap = data.currentWindow();
wrap.value().addPass(count);
}
这里的
data
是一个非常核心的成员，主要用于判断将数据放到那个窗口（桶）中，
wrap.value().addPass(count)
负责将数据写入
这个
BucketLeapArray
就实现了滑动窗口
当调用
data.currentWindow()
寻找当前该写入的windows时，最终会调用以下方法，这个是非常关键的方法
public WindowWrap<T> currentWindow(long timeMillis) {
if (timeMillis < 0) {
return null;
}
//1. 根据当前的诗句计算数组的Index
int idx = calculateTimeIdx(timeMillis);
// Calculate current bucket start time.
//2. 由于是相当于是环形数组，需要计算一下窗口开始时间，用于复用窗口时覆盖用
long windowStart = calculateWindowStart(timeMillis);
/*
* Get bucket item at given time from the array.
*
* (1) Bucket is absent, then just create a new bucket and CAS update to circular array.
* (2) Bucket is up-to-date, then just return the bucket.
* (3) Bucket is deprecated, then reset current bucket and clean all deprecated buckets.
*/
while (true) {
WindowWrap<T> old = array.get(idx);
if (old == null) {
//如果进到这里，说明刚启动不久，还有桶还没创建，新建一个
/*
* B0 B1 B2 NULL B4
* ||_______|_______|_______|_______|_______||___
* 200 400 600 800 1000 1200 timestamp
* ^
* time=888
* bucket is empty, so create new and update
*
* If the old bucket is absent, then we create a new bucket at {@code windowStart},
* then try to update circular array via a CAS operation. Only one thread can
* succeed to update, while other threads yield its time slice.
*/
WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
if (array.compareAndSet(idx, null, window)) {
// 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()) {
//进到这里来，说明还在同一个桶的时间区间内（start相同），直接返回
/*
* B0 B1 B2 B3 B4
* ||_______|_______|_______|_______|_______||___
* 200 400 600 800 1000 1200 timestamp
* ^
* time=888
* startTime of Bucket 3: 800, so it's up-to-date
*
* If current {@code windowStart} is equal to the start timestamp of old bucket,
* that means the time is within the bucket, so directly return the bucket.
*/
return old;
} else if (windowStart > old.windowStart()) {
// 进到这里来，说明已经在数据中转了一圈，复用的是旧的窗口，需要重置下旧窗口
/*
* (old)
* B0 B1 B2 NULL B4
* |_______||_______|_______|_______|_______|_______||___
* ... 1200 1400 1600 1800 2000 2200 timestamp
* ^
* time=1676
* startTime of Bucket 2: 400, deprecated, should be reset
*
* If the start timestamp of old bucket is behind provided time, that means
* the bucket is deprecated. We have to reset the bucket to current {@code windowStart}.
* Note that the reset and clean-up operations are hard to be atomic,
* so we need a update lock to guarantee the correctness of bucket update.
*
* The update lock is conditional (tiny scope) and will take effect only when
* bucket is deprecated, so in most cases it won't lead to performance loss.
*/
if (updateLock.tryLock()) {
try {
// Successfully get the update lock, now we reset the bucket.
return resetWindowTo(old, windowStart);
} finally {
updateLock.unlock();
}
} else {
// Contention failed, the thread will yield its time slice to wait for bucket available.
Thread.yield();
}
} else if (windowStart < old.windowStart()) {
// Should not go through here, as the provided time is already behind.
return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
}
}
}
我们的demo是创建了一个有60个桶的BucketLeapArray，就这样一直顺序向后循环使用。
定义桶结构的类为
MetricBucket
public class MetricBucket {
private final LongAdder[] counters;
...
public MetricBucket() {
MetricEvent[] events = MetricEvent.values();
this.counters = new LongAdder[events.length];
for (MetricEvent event : events) {
counters[event.ordinal()] = new LongAdder();
}
...
}
public MetricBucket add(MetricEvent event, long n) {
counters[event.ordinal()].add(n);
return this;
}
}
