标签：node count sentinel resourceWrapper 限流 context Sentinel entry 源码
1. git clone senetinel 源码到本地，切换到release1.8分支
2. 找到FlowQpsDemo.java， 根据sentinel自带的案例来学习sentinel的原理
3. 先看main方法
public static void main(String[] args) throws Exception {
initFlowQpsRule();
//tick();
// first make the system run on a very low condition
simulateTraffic();
System.out.println("===== begin to do flow control");
System.out.println("only 20 requests per second can pass");
}
这里首先初始化了限流规则，然后是调用模拟流量，先看initFlowQpsRule
private static void initFlowQpsRule() {
List<FlowRule> rules = new ArrayList<FlowRule>();
FlowRule rule1 = new FlowRule();
rule1.setResource(KEY);
// set limit qps to 20
rule1.setCount(20);
rule1.setGrade(RuleConstant.FLOW_GRADE_QPS);
rule1.setLimitApp("default");
rules.add(rule1);
FlowRuleManager.loadRules(rules);
}
这里通过设置 资源名称 和他对应的rule来绑定他们之间的映射关系，完成初始化，全局唯一
4. 再看simulateTraffic
private static void simulateTraffic() {
for (int i = 0; i < threadCount; i++) {
Thread t = new Thread(new RunTask());
t.setName("simulate-traffic-Task");
t.start();
}
}
这里启动了多条线程，来执行RunTask，runTask内部开始使用到了sentinel的API
static class RunTask implements Runnable {
@Override
public void run() {
while (!stop) {
Entry entry = null;
try {
entry = SphU.entry(KEY);
// token acquired, means pass
....
} catch (BlockException e1) {
block.incrementAndGet();
} catch (Exception e2) {
// biz exception
} finally {
...
if (entry != null) {
entry.exit();
}
}
....
}
}
}
在调用SphU.entry(KEY)时，会调用sentinel的规则调用链，来计算是否可以允许执行后面的步骤，如果不允许，将直接抛出异常，如果是触发限流，将会抛出BlockException
通过debug 可以看到如下流程图：
从流程图可以看出，核心是entryWithPriority
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
throws BlockException {
Context context = ContextUtil.getContext();
if (context instanceof NullContext) {
// The {@link NullContext} indicates that the amount of context has exceeded the threshold,
// so here init the entry only. No rule checking will be done.
return new CtEntry(resourceWrapper, null, context);
}
if (context == null) {
// Using default context.
context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
}
// Global switch is close, no rule checking will do.
if (!Constants.ON) {
return new CtEntry(resourceWrapper, null, context);
}
// 创建调用链，一个资源名称绑定同一个调用链
ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
/*
* Means amount of resources (slot chain) exceeds {@link Constants.MAX_SLOT_CHAIN_SIZE},
* so no rule checking will be done.
*/
if (chain == null) {
return new CtEntry(resourceWrapper, null, context);
}
Entry e = new CtEntry(resourceWrapper, chain, context);
try {
// 核心函数
chain.entry(context, resourceWrapper, null, count, prioritized, args);
} catch (BlockException e1) {
e.exit(count, args);
throw e1;
} catch (Throwable e1) {
// This should not happen, unless there are errors existing in Sentinel internal.
RecordLog.info("Sentinel unexpected exception", e1);
}
return e;
}
# Sentinel default ProcessorSlots
com.alibaba.csp.sentinel.slots.nodeselector.NodeSelectorSlot
com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot
com.alibaba.csp.sentinel.slots.logger.LogSlot
com.alibaba.csp.sentinel.slots.statistic.StatisticSlot
com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot
com.alibaba.csp.sentinel.slots.system.SystemSlot
com.alibaba.csp.sentinel.slots.block.flow.FlowSlot
com.alibaba.csp.sentinel.slots.block.degrade.DegradeSlot
其中限流我们主要关注 StatisticSlot 和 FlowSlot，StatisticSlot用于统计在单位统计时间内的流量情况，FlowSlot用于根据rule 和流量情况判断是否运行通行
先看FlowSlot的源码,调用链会调用到他的entry方法
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
//这里检查是否可通行，不可将抛出异常，不往下走了
checkFlow(resourceWrapper, context, node, count, prioritized);
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
void checkFlow(ResourceWrapper resource, Context context, DefaultNode node, int count, boolean prioritized)
throws BlockException {
checker.checkFlow(ruleProvider, resource, context, node, count, prioritized);
}
由流程图克制，最终是调用DefaultController.canPass来判断
@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
int curCount = avgUsedTokens(node);
if (curCount + acquireCount > count) {
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;
}
private int avgUsedTokens(Node node) {
if (node == null) {
return DEFAULT_AVG_USED_TOKENS;
}
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}
通过avgUsedToken访问node（这里debug可知是ClusterNode，因为是default模式），去拿statisticsNode逻辑中缓存在node中的Qps数据，即单位时长内的流量值；这里面有一个重要的数据结构LeapArray<MetricBucket> , 是sentinel设计的一个 环形数组用于流量统计，将在下节讲解
到此FlowSlot的统计流程就讲解完毕，即每个请求进来，成功后都会缓存在对应的Node中进行计数，在FlowSlot中判断单位时间内是否已达到上限，达到则抛出异常，阻止继续向下，（或者调用配置好的handler执行）
标签：node,
count,
sentinel,
resourceWrapper,
限流,
context,
Sentinel,
entry,
源码
From： https://www.cnblogs.com/gradyblog/p/16584923.html