之前发过一篇文章，介绍了alibaba Sentinel限流功能。Alibaba Sentinel限流功能
限流依赖的基础就是一个基于滑动时间窗口的计数器。
固定时间窗口
介绍滑动时间窗口前，先简单介绍下固定时间窗口，见下图
以统计QPS为例，我们可以将时间按照固定间隔进行切分，比如1000ms（一秒），统计每一个时间窗口内的计数，然后得出QPS，这是一种最简单的统计方式。
那么这种方式的缺点是什么呢?
比如我们定义了规则，QPS不能超过一万，如果在900ms和1100 ms分别进来一万流量，显然是满足流控限制的，但实际上，这个流量已经是两万QPS了。我们的流控限制在这种固定时间窗口下，起不到应有的限流作用，会导致服务过载。
滑动时间窗口
为了规避这个问题，滑动时间窗口将时间切分为多个窗口（一般是两个），窗口指针随着时间往后滑动，见下图
如图，滑动时间窗口将每秒钟时间（1000ms）切分为2个窗口，W1永远指向前半秒（前500ms），w2永远指向后半秒（后500ms），初始时，W1指向[0,500)ms区间，W2指向[500,1000)ms区间，当时间往后走到1001ms时，w1会去指向[1000,1500)ms区间。W2同理，会不断的替换为新的区间，以实现窗口滑动。用W1+W2两个区间的计数之和进行QPS计算，这样就解决了固定时间窗口下的临界流量问题。
Alibaba Sentinel代码实现
Sentinel实现滑动时间窗口，基于的是类OccupiableBucketLeapArray，其特殊的点就是，这个数据结构除了持有两个正常的时间窗口之外，还持有一个完全相同结构的borrowArray，其中包含两个未来的时间窗口。后续将介绍这个特殊结构的作用。
其主要逻辑在父类LeapArray中，实现时间窗口初始化，获取，滑动。
根据当前系统时间戳，去获取归属的时间窗口，主要逻辑包含下述注释中的三步
1.新建时间窗口 2.命中时间窗口 3.时间窗口起始值更新
WindowWrap<T> old = array.get(idx);
//1.如果老的时间窗口不存在，则新建新的时间窗口，通过cas的方式进行替换
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();
}
//2.如果时间戳在窗口之内，则直接返回
} else if (windowStart == old.windowStart()) {
/*
* 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;
//3.如果时间戳已经大于老窗口，则将老窗口的时间指向新的起始值
} 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);
基于以上的滑动时间窗口，限流的具体过程见注释1，2，3
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
//1.计算当前窗口计数之和
int curCount = avgUsedTokens(node);
//2.比较当前流量与规则限制
if (curCount + acquireCount > count) {
//3.即使超限，如果prioritized设为true，则认为是重要业务，可以尝试让业务线程sleep到下一个窗口，借用下一个窗口的计数
if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {
long currentTime;
long waitInMs;
currentTime = TimeUtil.currentTimeMillis();
waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);
if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {
node.addWaitingRequest(currentTime + waitInMs, acquireCount);
node.addOccupiedPass(acquireCount);
sleep(waitInMs);
// PriorityWaitException indicates that the request will pass after waiting for {@link @waitInMs}.
throw new PriorityWaitException(waitInMs);
}
}
return false;
}
return true;
}
前两个步骤都很好理解，如果定义了限流规则为一万QPS，当流量超限，不让通过即可，不允许访问服务。
可是如果是重要业务，超限了直接失败显然不行，Sentinel除了上图的两个window，还特意引入了一个包含两个未来时间窗口的borrowArray，先借用未来的计数，给与业务通过，同时让业务线程sleep一段时间，去落在新窗口上，而且当时间滑动到新的窗口时，也不用新建一个空计数的window，直接使用这个borrowArray中window的计数。这也是前文提到OccupiableBucketLeapArray特殊数据结构的作用。
public MetricBucket newEmptyBucket(long time) {
MetricBucket newBucket = new MetricBucket();
MetricBucket borrowBucket = borrowArray.getWindowValue(time);
if (borrowBucket != null) {
newBucket.reset(borrowBucket);
}
return newBucket;
}
