前言
接上文， 在了解了Sentinel如何进行流量统计之后，我们就可以来看看Sentinel是如何完成限流操作的了。根据之前的描述，我们先还是来看下整个Slot Chain流程图：
Slot Chain 调用链
从上图中我们可以看到，限流操作应该是在紧跟StatisticSlot的FlowSlot中完成的。那我们就去FlowSlot中一探究竟。
流量控制
下图是流量控制所在的代码组织：
FlowSlot 所在
按惯例，我们先来看看FlowSlot中的entry方法：
public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, Object... args)
throws Throwable {
//检查是否能够限流通过
FlowRuleManager.checkFlow(resourceWrapper, context, node, count);
//调用责任链下游的Slot的entry
fireEntry(context, resourceWrapper, node, count, args);
}
@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
fireExit(context, resourceWrapper, count, args);
}
}
这段代码很简单，我们重定向到FlowRuleManager的checkFlow方法:
public static void checkFlow(ResourceWrapper resource, Context context, DefaultNode node, int count)
throws BlockException {
//获取调用的resource对应的所有限流规则
List<FlowRule> rules = flowRules.get(resource.getName());
if (rules != null) {
for (FlowRule rule : rules) {
//逐个规则判断是否触发限流操作
if (!rule.passCheck(context, node, count)) {
//如果passCheck失败，说明触发限流，直接抛出FlowException
throw new FlowException(rule.getLimitApp());
}
}
}
}
我们马不停蹄再到FlowRule的passCheck:
@Override
public boolean passCheck(Context context, DefaultNode node, int acquireCount, Object... args) {
//是否有调用方需要被限流，默认情况下limitApp为'default'
String limitApp = this.getLimitApp();
//正常限流不会触发这种情况
if (limitApp == null) {
return true;
}
// 获得上下文的调用方，这里context会在后面的文章中分析
String origin = context.getOrigin();
//根据调用方和上下问以及FlowRule所配置的Strategy来获取应该用于限流的统计Node
//这个Node可以参照上文所描述的StatisticNode
Node selectedNode = selectNodeByRequesterAndStrategy(origin, context, node);
//如果没有合乎规则的Node，则直接返回true，表示通过
if (selectedNode == null) {
return true;
}
//如果存在统计Node，
//则通过controller来判断是否需要限流
//这个controller通过设置FlowRule的controllerBehavior来区分
//默认的实现有：0. default, 1. warm up, 2. rate limiter
return controller.canPass(selectedNode, acquireCount);
}
通过上面代码的分析，我们知道真正的限流逻辑藏在了FlowRule的controller里面，而这个controller有三种实现，我们就挨个来看
Default 默认方式
@Override
public boolean canPass(Node node, int acquireCount) {
//获取已经使用过的令牌
int curCount = avgUsedTokens(node);
//如果使用过的令牌数目加上这次的超过了限流的数目
//则返回false，表示不能通过
if (curCount + acquireCount > count) {
return false;
}
//否则返回true
return true;
}
private int avgUsedTokens(Node node) {
//一般无法触发
if (node == null) {
return -1;
}
//如果按照并发线程数限流则返回统计中的线程数目
//否则就返回现在已经pass的Qps数目
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)node.passQps();
}
这段代码很简单了，逻辑也都在注释中，值得注意的是两点：
- 这里没有用任何加锁操作，这里我跟开发团队交流得到的结论是这里为了性能尽可能的实现了lock free，但是会导致在临界状态下多放，因为从这次返回true到entry调用回到StatisticSlot中addPass会有一个时间窗口，在加上多线程的环境自然就会有竞态问题。不过这相当于是性能与正确性的一个trade-off，大家自己可以用Sentinel提供的demo测试下，在32线程并发下偶尔会有多放，但是也确实是能够接受的。
- 这里的QPS限流与我之前文章所描述的令牌桶算法也有区别，这里相当于是在当前时间一开始就把QPS个令牌全部放出，而不是每一个1000/QPS个毫秒间隔释放一个令牌，这也是一种操作上的简化。
Warm Up 冷启动方式
根据文档描述：
该方式主要用于系统长期处于低水位的情况下，当流量突然增加时，直接把系统拉升到高水位可能瞬间把系统压垮。通过"冷启动"，让通过的流量缓慢增加，在一定时间内逐渐增加到阈值上限，给冷系统一个预热的时间，避免冷系统被压垮的情况。
根据注释，Sentinel的预热限流实现参照了Guava的SmoothWarmUp 限流器的算法，我在这篇文章中对SmoothWarmUp的所用的算法和实现做过介绍，不清楚的同学请先移步，这篇文章就不再叙述了。
我们来看看代码：
@Override
public boolean canPass(Node node, int acquireCount) {
long passQps = node.passQps();
long previousQps = node.previousPassQps();
//装填令牌桶
syncToken(previousQps);
// 开始计算它的斜率
// 如果进入了警戒线，开始调整他的qps
long restToken = storedTokens.get();
if (restToken >= warningToken) {
long aboveToken = restToken - warningToken;
// 消耗的速度要比warning快，但是要比慢
// current interval = restToken*slope+1/count
//这里相当于计算了一个在警戒线以上的interval
//计算依据可以参考guava的依靠斜率来计算梯形面积部分代码
//1.0/count 为恒定速率
//有了interval 那么1.0/interval即为这时候期望的QPS
double warningQps = Math.nextUp(1.0 / (aboveToken * slope + 1.0 / count));
//拿这个QPS来做比较
if (passQps + acquireCount <= warningQps) {
return true;
}
} else {
//如果在警戒线下，就退化为基本的QPS流控
if (passQps + acquireCount <= count) {
return true;
}
}
return false;
}
private void syncToken(long passQps) {
long currentTime = TimeUtil.currentTimeMillis();
//获取当前秒
currentTime = currentTime - currentTime % 1000;
//获取上一次装填令牌桶的时间
long oldLastFillTime = lastFilledTime.get();
//如果当前秒数小于等于上次装填时间
//则return，表示一秒钟内只需要装填一次
if (currentTime <= oldLastFillTime) {
return;
}
//获取目前令牌桶中的令牌数目
long oldValue = storedTokens.get();
//根据当前时间和前一秒pass的Qps来装填
long newValue = coolDownTokens(currentTime, passQps);
//如果CAS成功，那做下一步操作
//就算冲突也无所谓，保证一秒装填一次就行
if (storedTokens.compareAndSet(oldValue, newValue)) {
//将前一秒的QPS减去
//注意点：如果没有前面保证一秒钟内只修改一次的策略，会有多减的可能性
long currentValue = storedTokens.addAndGet(0 - passQps);
//不要成为负数
if (currentValue < 0) {
storedTokens.set(0L);
}
//修改装填时间
lastFilledTime.set(currentTime);
}
}
private long coolDownTokens(long currentTime, long passQps) {
long oldValue = storedTokens.get();
long newValue = oldValue;
// 添加令牌的判断前提条件:
// 当令牌的消耗程度远远低于警戒线的时候
//这里将装填的速率恒定为1.0/QPS
//那么装填后的值即为oldValue + time / (1000 / count)
if (oldValue < warningToken) {
newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);
} else if (oldValue > warningToken) {
//如果之前的令牌桶数目已经高于警戒线，那么看pass了的QPS是否小于一个阈值，决定是否装填
if (passQps < (int)count / coldFactor) {
newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);
}
}
return Math.min(newValue, maxToken);
}
一些基础的变量与Guava实现的对应关系如下：
- warningToken 对应 thresholdPermits
- maxToken 对应 maxPermits
- slop 不变
计算方式也与Guava实现一致。大家需要注意的就是canPass中首先通过syncToken重新装填了一波令牌桶，并且把前一秒的Qps减掉，然后再根据当前令牌桶中剩余的来计算是否需要限流。这一块可能需要大家多理解一下Guava的算法就清楚了。
Rate Limiter(PaceController) 匀速器
这里作者使用了漏桶算法来完成匀速器的功能，主要是做到了一个流量整形的功能，这样的功能非常适合处理突发性流量的处理，比方说高并发整形到匀速队列中，我们来看看代码：
@Override
public boolean canPass(Node node, int acquireCount) {
// 按照斜率来计算计划中应该什么时候通过（源代码注释）
// 这里获取当前时间
long currentTime = TimeUtil.currentTimeMillis();
//获取acquireCount个令牌需要的时间
long costTime = Math.round(1.0 * (acquireCount) / count * 1000);
//期待时间(源代码注释)
//上次pass的时间加上这次获取需要花费的时间就得到了期望的时间
long expectedTime = costTime + latestPassedTime.get();
//如果期望时间小于等于当前时间
//说明现在立马就能够获得令牌
if (expectedTime <= currentTime) {
//这里会有冲突,然而冲突就冲突吧.（源代码注释）
latestPassedTime.set(currentTime);
return true;
} else {
//如果期望时间大于当前时间
// 计算自己需要的等待时间（源代码注释）
long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
//如果等待的时间大于等于最大的队列等待时
if (waitTime >= maxQueueingTimeMs) {
//返回false标明这一次需要被限流
return false;
} else {
//这里应用了一个类似于double check的做法
//先调用latestPassedTime这个AtomicLong的原子加
//并获取原子加后的新值（确实不知道为啥叫old）
long oldTime = latestPassedTime.addAndGet(costTime);
try {
//再做一次判断，如果这时候有并发的原子加操作
//那么总有线程得到的oldTime与之前期望的不一致
//那么在计算一次waitTime
waitTime = oldTime - TimeUtil.currentTimeMillis();
//如果这时候waitTime超过了最长队列等待时间
if (waitTime >= maxQueueingTimeMs) {
//把加上去的时间又给减回去
//并且返回false，触发限流
latestPassedTime.addAndGet(-costTime);
return false;
}
//如果到达了这里，恭喜你，你已经通过了匀速器的考验
//现在需要做的就是等待该等待的时间，然后返回false
//完成匀速器的使命
Thread.sleep(waitTime);
return true;
} catch (InterruptedException e) {
}
}
}
//正常情况不该到这儿，返回false
return false;
}
大致的代码的意义我也在注释中加以说明，这里通过一些比较巧妙的原子操作和双重检查来完成了匀速器的一个功能。
结语
测试部分我这里就不再贴结果了，Sentinel提供了比较完备的demo来测试各种情况，代码在这里，有疑问的同学可以自行运行：
测试代码所在
本篇文章通过对Sentinel提供的三种限流方案做了代码分析，希望大家能够喜欢！
同时我这里也做一点点猜想，对于集群限流，是否可以不动用统计代码，只是在流控中将本机的数目上传集中存储（例如redis），并拉取整个集群的流量数据来本地做限流？这样可以尽可能较少代码量，并保证功能，当然按照Sentinel的roadmap，11月份应该就会提供集群限流功能，我也很期待到时候的实现会怎样？
网友评论