(
chainMap.size() + 1);
newMap.putAll(chainMap);
newMap.put(resourceWrapper, chain);
chainMap = newMap;
}
}
}
return chain;
}
- 最终调用
DefaultSlotChainBuilder.build() 方法构造 DefaultProcessorSlotChain；
@Override
public ProcessorSlotChain build() {
ProcessorSlotChain chain = new DefaultProcessorSlotChain();
List sortedSlotList = SpiLoader.of(ProcessorSlot.class).loadInstanceListSorted();
for (ProcessorSlot slot : sortedSlotList) {
if (!(slot instanceof AbstractLinkedProcessorSlot)) {
RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() + ") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
continue;
}
chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
}
return chain;
}
- 可以看到最后 ProcessorSlotChain 链中有 10 个 Slot 插槽：
- 在本篇笔记中我们关注 3 个槽：
- FlowSlot：进行流控规则校验，对应本篇《3. 流控槽实施流控逻辑》；
- StatisticSlot：实现指标数据的统计，对应本篇《4. 统计槽实施指标数据统计》；
- DegradeSlot：服务熔断，对应本篇《5. 熔断槽实施服务熔断》
2.2.2 操作 Slot 槽的入口
- 操作 Slot 槽的入口方法是：
ProcessorSlot.entry()；
- 接着会以遍历链表的方式操作每个 Slot 槽，其中就有：FlowSlot、StatisticSlot、DegradeSlot 等。分别对应下面的《3. 流控槽实施流控逻辑》、《4. 统计槽实施指标数据统计》和《5. 熔断槽实施服务熔断》；
3. 流控槽实施流控逻辑 FlowSlot.entry()
- 进入
ProcessorSlot.entry() 方法，它会遍历每个 Slot 插槽，并对其进行操作，其中会经过 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);
//调用下一个 Slot
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
- 进入
FlowSlot.checkFlow() 方法，最终调用 FlowRuleChecker.checkFlow() 方法，方法的逻辑和源码如下：
- 遍历所有流控规则 FlowRule；
- 针对每个规则调用 canPassCheck 进行校验；
public void checkFlow(Function> ruleProvider, ResourceWrapper resource,
Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
if (ruleProvider == null || resource == null) {
return;
}
//【断点步入 3.1】获取流控规则
Collection rules = ruleProvider.apply(resource.getName());
if (rules != null) {
//遍历所有流控规则 FlowRule
for (FlowRule rule : rules) {
//【点进去 3.2】校验每条规则
if (!canPassCheck(rule, context, node, count, prioritized)) {
throw new FlowException(rule.getLimitApp(), rule);
}
}
}
}
3.1 获取流控规则 FlowSlot.ruleProvider.apply()
- 进入
FlowSlot.ruleProvider.apply() 方法，获取到 Sentinel 控制台上的流控规则；
private final Function> ruleProvider = new Function>() {
@Override
public Collection apply(String resource) {
// Flow rule map should not be null.
Map> flowRules = FlowRuleManager.getFlowRuleMap();
return flowRules.get(resource);
}
};
3.2 校验每条规则 FlowRuleChecker.canPassCheck()
- 进入
FlowRuleChecker.canPassCheck() 方法，分集群和单机模式校验每条规则；
public boolean canPassCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node, int acquireCount, boolean prioritized) {
String limitApp = rule.getLimitApp();
if (limitApp == null) {
return true;
}
//集群模式
if (rule.isClusterMode()) {
return passClusterCheck(rule, context, node, acquireCount, prioritized);
}
//【点进去】单机模式
return passLocalCheck(rule, context, node, acquireCount, prioritized);
}
- 由于我们是单机模式，进入
FlowRuleChecker.passLocalCheck() 方法，其主要逻辑和源码如下：
- 根据来源和策略获取 Node，从而拿到统计的 runtime 信息；
- 使用流量控制器检查是否让流量通过；
private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount, boolean prioritized) {
//【点进去 3.2.1】获取 Node
Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);
if (selectedNode == null) {
return true;
}
//【点进去 3.2.2】获取流控的处理策略
return rule.getRater().canPass(selectedNode, acquireCount, prioritized);
}
3.2.1 获取 Node FlowRuleChecker.selectNodeByRequesterAndStrategy()
- 进入
FlowRuleChecker.selectNodeByRequesterAndStrategy() 方法，其根据 FlowRule 中配置的 Strategy 和 limitApp 属性，返回不同处理策略的 Node；
static Node selectNodeByRequesterAndStrategy(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node) {
//limitApp 不能为空
String limitApp = rule.getLimitApp();
int strategy = rule.getStrategy();
String origin = context.getOrigin();
//场景1：限流规则设置了具体应用，如果当前流量就是通过该应用的，则命中场景1
if (limitApp.equals(origin) && filterOrigin(origin)) {
if (strategy == RuleConstant.STRATEGY_DIRECT) {
// Matches limit origin, return origin statistic node.
return context.getOriginNode();
}
return selectReferenceNode(rule, context, node);
} else if (RuleConstant.LIMIT_APP_DEFAULT.equals(limitApp)) {
//场景2：限流规则未指定任何具体应，默认为default，则当前流量直接命中场景2
if (strategy == RuleConstant.STRATEGY_DIRECT) {
// Return the cluster node.
return node.getClusterNode();
}
return selectReferenceNode(rule, context, node);
} else if (RuleConstant.LIMIT_APP_OTHER.equals(limitApp) && FlowRuleManager.isOtherOrigin(origin, rule.getResource())) {
//场景3：限流规则设置的是other，当前流量未命中前两种场景
if (strategy == RuleConstant.STRATEGY_DIRECT) {
return context.getOriginNode();
}
return selectReferenceNode(rule, context, node);
}
return null;
}
- 假设我们对接口 UserService 配置限流 1000 QPS，这 3 种场景分别如下：
- 场景 1：目的是优先保障重要来源的流量。我们需要区分调用来源，将限流规则细化。对A应用配置500QPS，对B应用配置200QPS，此时会产生两条规则：A应用请求的流量限制在500，B应用请求的流量限制在200；
- 场景 2：没有特别重要来源的流量。我们不想区分调用来源，所有入口调用 UserService 共享一个规则，所有 client 加起来总流量只能通过 1000 QPS；
- 场景 3：配合第1种场景使用，在长尾应用多的情况下不想对每个应用进行设置，没有具体设置的应用都将命中；
3.2.2 获取流控的处理策略 `FlowRule.getRater().canPass()
- 进入
FlowRule.getRater().canPass() 方法，首先通过 FlowRule.getRater() 获得流控行为 TrafficShapingController，这是一个接口，有四种实现类，如下图所示：
- 有以下四种处理策略：
- DefaultController：直接拒绝；
- RateLimiterController：匀速排队；
- WarmUpController：冷启动（预热）；
- WarmUpRateLimiterController：匀速+冷启动。
- 最终调用
TrafficShapingController.canPass() 方法，执行流控行为；
4. 统计槽实施指标数据统计 StatisticSlot.entry()
- 限流的核心是限流算法的实现，Sentinel 默认采用滑动窗口算法来实现限流，具体的指标数据统计由 StatisticSlot 实现；
- 我们给
StatisticSlot.entry() 方法里的语句打上断点，运行到光标处；
StatisticSlot.entry() 方法的核心是使用 Node 统计“增加线程数”和“请求通过数”；
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
try {
//先执行后续 Slot 检查，再统计数据（即先调用后续所有 Slot）
fireEntry(context, resourceWrapper, node, count, prioritized, args);
//【断点步入】使用 Node 统计“增加线程数”和“请求通过数”
node.increaseThreadNum();
node.addPassRequest(count);
//如果存在来源节点，则对来源节点增加线程数和请求通过数
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseThreadNum();
context.getCurEntry().getOriginNode().addPassRequest(count);
}
//如果是入口流量，则对全局节点增加线程数和请求通过数
if (resourceWrapper.getEntryType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseThreadNum();
Constants.ENTRY_NODE.addPassRequest(count);
}
//执行事件通知和回调函数
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onPass(context, resourceWrapper, node, count, args);
}
//处理优先级等待异常
} catch (PriorityWaitException ex) {
node.increaseThreadNum();
//如果有来源节点，则对来源节点增加线程数
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseThreadNum();
}
//如果是入口流量，对全局节点增加线程数
if (resourceWrapper.getEntryType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseThreadNum();
}
//执行事件通知和回调函数
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onPass(context, resourceWrapper, node, count, args);
}
//处理限流、熔断等异常
} catch (BlockException e) {
//省略
throw e;
//处理业务异常
} catch (Throwable e) {
context.getCurEntry().setError(e);
throw e;
}
}
4.1 统计“增加线程数”和“请求通过数”
- 这两个方法都是调用同一个类的，笔者以第一个为例，进入
DefaultNode.increaseThreadNum() 方法，最终调用的是 StatisticNode.increaseThreadNum()，而统计也是依靠 StatisticNode 维护的，这里放上 StatisticNode 的统计核心与源码：
- StatisticNode 持有两个计数器 Metric 对象，统计行为是通过 Metric 完成的；
public class StatisticNode implements Node {
//省略其他代码
//【断点步入】最近 1s 滑动窗口计数器（默认 1s）
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
IntervalProperty.INTERVAL);
//最近 1min 滑动窗口计数器（默认 1min）
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
//增加 “请求通过数”
@Override
public void addPassRequest(int count) {
rollingCounterInSecond.addPass(count);
rollingCounterInMinute.addPass(count);
}
//增加 RT 和成功数
@Override
public void addRtAndSuccess(long rt, int successCount) {
rollingCounterInSecond.addSuccess(successCount);
rollingCounterInSecond.addRT(rt);
rollingCounterInMinute.addSuccess(successCount);
rollingCounterInMinute.addRT(rt);
}
//增加“线程数”
@Override
public void increaseThreadNum() {
curThreadNum.increment();
}
}
- 这里还有减少请求通过数（线程数）、统计最大值等方法，由于篇幅有限，这里不放出，感兴趣的读者可以自己 DeBug 进入看看；
4.2 数据统计的数据结构
4.2.1 ArrayMetric 指标数组
- ArrayMetric 的构造方法需要先给方法打上断点，重新 DeBug，在初始化时注入构造；
public class ArrayMetric implements Metric {
//省略其他代码
//【点进去 4.2.2】数据存储
private final LeapArray data;
//最近 1s 滑动计数器用的是 OccupiableBucketLeapArray
public ArrayMetric(int sampleCount, int intervalInMs) {
this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
}
//最近 1min 滑动计数器用的是 BucketLeapArray
public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy) {
if (enableOccupy) {
this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
} else {
this.data = new BucketLeapArray(sampleCount, intervalInMs);
}
}
//增加成功数
@Override
public void addSuccess(int count) {
WindowWrap wrap = data.currentWindow();
wrap.value().addSuccess(count);
}
//增加通过数
@Override
public void addPass(int count) {
WindowWrap wrap = data.currentWindow();
wrap.value().addPass(count);
}
//增加 RT
@Override
public void addRT(long rt) {
WindowWrap wrap = data.currentWindow();
wrap.value().addRT(rt);
}
}
4.2.2 LeapArray 环形数组
- LeapArray 是处理数据的核心数据结构，采用滑动窗口算法；
- ArrayMetric 中持有 LeapArray 对象，所有方法都是对 LeapArray 进行操作；
- LeapArray 是环形的数据结构，为了节约内存，它存储固定个数的窗口对象 WindowWrap，只保存最近一段时间的数据，新增的时间窗口会覆盖最早的时间窗口；
public abstract class LeapArray {
//省略其他代码
//单个窗口的长度（1个窗口多长时间）
protected int windowLengthInMs;
//采样窗口个数
protected int sampleCount;
//全部窗口的长度（全部窗口多长时间）
protected int intervalInMs;
private double intervalInSecond;
//窗口数组：存储所有窗口（支持原子读取和写入）
protected final AtomicReferenceArray> array;
//更新窗口数据时用的锁
private final ReentrantLock updateLock = new ReentrantLock();
public LeapArray(int sampleCount, int intervalInMs) {
//计算单个窗口的长度
this.windowLengthInMs = intervalInMs / sampleCount;
this.intervalInMs = intervalInMs;
this.intervalInSecond = intervalInMs / 1000.0;
this.sampleCount = sampleCount;
this.array = new AtomicReferenceArray<>(sampleCount);
}
//【点进去 4.2.3】获取当前窗口
public WindowWrap currentWindow() {
//这里参数是当前时间
return currentWindow(TimeUtil.currentTimeMillis());
}
//获取指定时间的窗口
public WindowWrap currentWindow(long timeMillis) {
if (timeMillis < 0) {
return null;
}
// 计算数组下标
int idx = calculateTimeIdx(timeMillis);
//计算当前请求对应的窗口开始时间
long windowStart = calculateWindowStart(timeMillis);
/*
* 从 array 中获取窗口。有 3 种情况：
* (1) array 中窗口不在，创建一个 CAS 并写入 array；
* (2) array 中窗口开始时间 = 当前窗口开始时间，直接返回；
* (3) array 中窗口开始时间 < 当前窗口开始时间，表示 o1d 窗口已过期，重置窗口数据并返回；
*/
while (true) {
// 取窗口
WindowWrap old = array.get(idx);
//（1）窗口不在
if (old == null) {
//创建一个窗口
WindowWrap window = new WindowWrap(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
//CAS将窗口写进 array 中并返回（CAS 操作确保只初始化一次）
if (array.compareAndSet(idx, null, window)) {
return window;
} else {
//并发写失败，释放 CPU 资源，避免有线程长时间占用 CPU，一般下次来的时候 array 中有数据了会命中第2种情况；
Thread.yield();
}
//（2）array 中窗口开始时间 = 当前窗口开始时间
} else if (windowStart == old.windowStart()) {
//直接返回
return old;
//（3）array 中窗口开始时间 < 当前窗口开始时间
} else if (windowStart > old.windowStart()) {
//尝试获取更新锁
if (updateLock.tryLock()) {
try {
//拿到锁的线程才重置窗口
return resetWindowTo(old, windowStart);
} finally {
//释放锁
updateLock.unlock();
}
} else {
//并发加锁失败，释放 CPU 资源，避免有线程长时间占用 CPU，一般下次来的时候因为 old 对象时间更新了会命中第 2 种情况；
Thread.yield();
}
//理论上不会出现
} else if (windowStart < old.windowStart()) {
// 正常情况不会进入该分支（机器时钟回拨等异常情况）
return new WindowWrap(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
}
}
}
//计算索引
private int calculateTimeIdx(/*@Valid*/ long timeMillis) {
//timeId 降低时间精度
long timeId = timeMillis / windowLengthInMs;
//计算当前索引，这样我们就可以将时间戳映射到 leap 数组
return (int)(timeId % array.length());
}
//计算窗口开始时间
protected long calculateWindowStart(/*@Valid*/ long timeMillis) {
return timeMillis - timeMillis % windowLengthInMs;
}
}
4.2.3 WindowWrap 窗口包装类
- WindowWrap 是一个窗口对象，它是一个包装类，包装的对象是 MetricBucket；
public class WindowWrap {
//窗口长度，与 LeapArray 的 windowLengthInMs 一致
private final long windowLengthInMs;
//窗口开始时间，其值是 windowLengthInMs 的整数倍
private long windowStart;
//窗口的数据，支持 MetricBucket 类型，存储统计数据
private T value;
//省略其他代码
}
4.2.4 MetricBucket 指标桶
- MetricBucket 类的定义如下，可以发现指标数据存在 LongAdder[] counters中；
- LongAdder 是 JDK1.8 中新增的类，用于在高并发场景下代替AtomicLong，以用空间换时间的方式降低了 CAS 失败的概率，从而提高性能；
public class MetricBucket {
/**
* 存储指标的计数器；
* LongAdder 是线程安全的计数器
* counters[0] PASS 通过数；
* counters[1] BLOCK 拒绝数；
* counters[2] EXCEPTION 异常数；
* counters[3] SUCCESS 成功数；
* counters[4] RT 响应时长；
* counters[5] OCCUPIED_PASS 预分配通过数;
**/
private final LongAdder[] counters;
//最小 RT，默认值是 5000ms
private volatile long minRt;
//构造中初始化
public MetricBucket() {
MetricEvent[] events = MetricEvent.values();
this.counters = new LongAdder[events.length];
for (MetricEvent event : events) {
counters[event.ordinal()] = new LongAdder();
}
initMinRt();
}
//覆盖指标
public MetricBucket reset(MetricBucket bucket) {
for (MetricEvent event : MetricEvent.values()) {
counters[event.ordinal()].reset();
counters[event.ordinal()].add(bucket.get(event));
}
initMinRt();
return this;
}
private void initMinRt() {
this.minRt = SentinelConfig.statisticMaxRt();
}
//重置指标为0
public MetricBucket reset() {
for (MetricEvent event : MetricEvent.values()) {
counters[event.ordinal()].reset();
}
initMinRt();
return this;
}
//获取指标，从 counters 中返回
public long get(MetricEvent event) {
return counters[event.ordinal()].sum();
}
//添加指标
public MetricBucket add(MetricEvent event, long n) {
counters[event.ordinal()].add(n);
return this;
}
public long pass() {
return get(MetricEvent.PASS);
}
public long block() {
return get(MetricEvent.BLOCK);
}
public void addPass(int n) {
add(MetricEvent.PASS, n);
}
public void addBlock(int n) {
add(MetricEvent.BLOCK, n);
}
//省略其他代码
}
4.2.5 各数据结构的依赖关系
4.2.6 LeapArray 统计数据的大致思路
- 创建一个长度为 n 的数组，数组元素就是窗口，窗口包装了 1 个指标桶，桶中存放了该窗口时间范围中对应的请求统计数据；
- 可以想象成一个环形数组在时间轴上向右滚动，请求到达时，会命中数组中的一个窗口，那么该请求的数据就会存到命中的这个窗口包含的指标桶中；
- 当数组转满一圈时，会回到数组的开头，而此时下标为 0 的元素需要重复使用，它里面的窗口数据过期了，需要重置，然后再使用。具体过程如下图：
5. 熔断槽实施服务熔断 DegradeSlot.entry()
- 服务熔断是通过 DegradeSlot 来实现的，它会根据用户配置的熔断规则和系统运行时各个 Node 中的统计数据进行熔断判断；
- 注意：熔断功能在 Sentinel-1.8.0 版本前后有较大变化；
- 我们给
DegradeSlot.entry() 方法里的语句打上断点，运行到光标处；
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
//【断点步入】熔断检查
performChecking(context, resourceWrapper);
//调用下一个 Slot
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
- 进入
DegradeSlot.performChecking() 方法，其逻辑与源码如下：
void performChecking(Context context, ResourceWrapper r) throws BlockException {
//根据 resourceName 获取断路器
List circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
if (circuitBreakers == null || circuitBreakers.isEmpty()) {
return;
}
//循环判断每个断路器
for (CircuitBreaker cb : circuitBreakers) {
//【点进去】尝试通过断路器
if (!cb.tryPass(context)) {
throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule());
}
}
}
5.1 继续或取消熔断功能
- 进入
AbstractCircuitBreaker.tryPass() 方法，当请求超时并且处于探测恢复（半开状态，HALF-OPEN 状态）失败时继续断路功能；
@Override
public boolean tryPass(Context context) {
//当前断路器状态为关闭
if (currentState.get() == State.CLOSED) {
return true;
}
if (currentState.get() == State.OPEN) {
//【点进去】对于半开状态，我们尝试通过
return retryTimeoutArrived() && fromOpenToHalfOpen(context);
}
return false;
}
- 进入
AbstractCircuitBreaker.fromOpenToHalfOpen() 方法，实现状态的变更；
protected boolean fromOpenToHalfOpen(Context context) {
//尝试将状态从 OPEN 设置为 HALF_OPEN
if (currentState.compareAndSet(State.OPEN, State.HALF_OPEN)) {
//状态变化通知
notifyObservers(State.OPEN, State.HALF_OPEN, null);
Entry entry = context.getCurEntry();
//在 entry 添加一个 exitHandler entry.exit() 时会调用
entry.whenTerminate(new BiConsumer() {
@Override
public void accept(Context context, Entry entry) {
//如果有发生异常，重新将状态设置为OPEN 请求不同通过
if (entry.getBlockError() != null) {
currentState.compareAndSet(State.HALF_OPEN, State.OPEN);
notifyObservers(State.HALF_OPEN, State.OPEN, 1.0d);
}
}
});
//此时状态已设置为HALF_OPEN正常通行
return true;
}
//熔断
return false;
}
- 上述讲解了：状态从 OPEN 变为 HALF_OPEN，HALF_OPEN 变为 OPEN；
- 但状态从 HALF_OPEN 变为 CLOSE 需要在正常执行完请求后，由 entry.exit() 调用
DegradeSlot.exit() 方法来改变状态；
5.2 请求失败，启动熔断
- 状态从 HALF_OPEN 变为 CLOSE 的实现方法在
DegradeSlot.exit()；
@Override
public void exit(Context context, ResourceWrapper r, int count, Object... args) {
Entry curEntry = context.getCurEntry();
//无阻塞异常
if (curEntry.getBlockError() != null) {
fireExit(context, r, count, args);
return;
}
//通过资源名获取断路器
List circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
//没有配置断路器，则直接放行
if (circuitBreakers == null || circuitBreakers.isEmpty()) {
fireExit(context, r, count, args);
return;
}
if (curEntry.getBlockError() == null) {
for (CircuitBreaker circuitBreaker : circuitBreakers) {
//【点进去】在请求完成时
circuitBreaker.onRequestComplete(context);
}
}
fireExit(context, r, count, args);
}
- 进入
ExceptionCircuitBreaker.onRequestComplete() 方法，其主要逻辑与源码如下：
- 请求失败比例与总请求比例加 1，用于判断后续是否超过阈值；
@Override
public void onRequestComplete(Context context) {
Entry entry = context.getCurEntry();
if (entry == null) {
return;
}
Throwable error = entry.getError();
//简单错误计数器
SimpleErrorCounter counter = stat.currentWindow().value();
if (error != null) {
//异常请求数加 1
counter.getErrorCount().add(1);
}
//总请求数加 1
counter.getTotalCount().add(1);
//【点进去】超过阈值时变更状态
handleStateChangeWhenThresholdExceeded(error);
}
- 进入
ExceptionCircuitBreaker.handleStateChangeWhenThresholdExceeded() 方法，变更状态；
private void handleStateChangeWhenThresholdExceeded(Throwable error) {
//全开则直接放行
if (currentState.get() == State.OPEN) {
return;
}
//半开状态
if (currentState.get() == State.HALF_OPEN) {
//检查请求
if (error == null) {
//发生异常，将状态从半开 HALF_OPEN 转为关闭 CLOSE
fromHalfOpenToClose();
} else {
//无异常，解开半开状态
fromHalfOpenToOpen(1.0d);
}
return;
}
//计算是否超过阈值
List counters = stat.values();
long errCount = 0;
long totalCount = 0;
for (SimpleErrorCounter counter : counters) {
errCount += counter.errorCount.sum();
totalCount += counter.totalCount.sum();
}
if (totalCount < minRequestAmount) {
return;
}
double curCount = errCount;
if (strategy == DEGRADE_GRADE_EXCEPTION_RATIO) {
//熔断策略为：异常比例
curCount = errCount * 1.0d / totalCount;
}
if (curCount > threshold) {
transformToOpen(curCount);
}
}
6. Sentinel 源码结构图小结
- SphU.entry()：核心逻辑的入口函数；
- CtSph.entryWithPriority()：获取 Slot 链，操作 Slot 槽；
- CtSph.lookProcessChain()：获取 ProcessorSlot 链；
- DefaultSlotChainBuilder.build()：构造 DefaultProcessorSlotChain 链（里面有 10 个 Slot 插槽）；
- ProcessorSlot.entry()：遍历 ProcessorSlot 链；
-
FlowSlot.entry()：遍历到 FlowSlot 槽，限流规则；
- FlowSlot.checkFlow()：检查流量规则；
- FlowRuleChecker.checkFlow()：使用检查器检查流量规则；
- FlowSlot.ruleProvider.apply()：获取流控规则；
- FlowRuleChecker.canPassCheck()：校验每条规则；
- FlowRuleChecker.passClusterCheck()：集群模式；
- FlowRuleChecker.passLocalCheck()：单机模式；
- FlowRuleChecker.selectNodeByRequesterAndStrategy()：获取 Node；
- FlowRule.getRater()：获得流控行为 TrafficShapingController；
- TrafficShapingController.canPass()：执行流控行为；
-
StatisticSlot.entry：遍历到 StatisticSlot 槽，统计数据；
- DefaultNode.increaseThreadNum()：统计“增加线程数”；
- StatisticNode.increaseThreadNum()：统计“请求通过数”；
- ArrayMetric.ArrayMetric()：初始化指标数组；
- LeapArray：环形数组；
- MetricBucket：指标桶；
- DefaultNode.addPassRequest()：统计“增加线程数”；
- StatisticNode.addPassRequest()：同上；
-
DegradeSlot.entry()：遍历到 DegradeSlot 槽，服务熔断；
- DegradeSlot.performChecking()：执行检查；
- DegradeRuleManager.getCircuitBreakers()：根据 resourceName 获取断路器；
- AbstractCircuitBreaker.tryPass()：继续或取消熔断功能；
- AbstractCircuitBreaker.fromOpenToHalfOpen()：尝试通过半开状态；
-
DegradeSlot.exit()：请求失败（超时），启动熔断；
- ExceptionCircuitBreaker.onRequestComplete()：在请求完成时操作；
- ExceptionCircuitBreaker.handleStateChangeWhenThresholdExceeded()：变更状态；
最后
新人制作，如有错误，欢迎指出，感激不尽！
欢迎关注公众号，会分享一些更日常的东西！
如需转载，请标注出处！
