【Sentinel】Sentinel原码分析

本文内容来自【黑马】Sentinel从使用到源码解读笔记，做了部分修改和补充

---

目录

Sentinel 基本概念

基本流程

Node

Entry

定义资源的两种方式

使用try-catch定义资源

使用注解标记资源

基于注解标记资源的实现原理

Context

什么是Context

Context的初始化

ContextUtil

ProcessorSlotChain执行流程

入口

NodeSelectorSlot

ClusterBuilderSlot

LogSlot

StatisticSlot

AuthoritySlot

SystemSlot

ParamFlowSlot

令牌桶

FlowSlot

滑动时间窗口

时间窗口请求量统计

滑动窗口QPS计算

DegradeSlot

触发断路器

---

Sentinel 基本概念

基本流程

在 Sentinel 里面，所有的资源都对应一个资源名称以及一个 Entry。Entry 可以通过对主流框架的适配自动创建，也可以通过注解的方式或调用 API 显式创建；每一个 Entry 创建的时候，同时也会创建一系列功能插槽（slot chain）。这些插槽有不同的职责:

1. NodeSelectorSlot 负责收集资源的路径，并将这些资源的调用路径，以树状结构存储起来，用于根据调用路径来限流降级；
2. ClusterBuilderSlot 用于存储资源的统计信息以及调用者信息，例如该资源的 RT, QPS, thread count 等等，这些信息将用作为多维度限流，降级的依据；
3. LogSlot 用于响应日志记录块异常，以提供用于故障排除的具体日志；
4. StatisticSlot 用于记录、统计不同纬度的 runtime 指标监控信息；
5. AuthoritySlot 根据配置的黑白名单和调用来源信息，来做黑白名单控制；
6. SystemSlot 通过系统的状态，例如 load1 等，来控制总的入口流量；
7. ParamFlowSlot 负责通过频繁（“热点”）参数进行流控制的处理器插槽；
8. FlowSlot 则用于根据预设的限流规则以及前面 slot 统计的状态，来进行流量控制；
9. DegradeSlot 通过统计信息以及预设的规则，来做熔断降级；

总体的框架如下（旧版，未更新）:

Sentinel 将 ProcessorSlot 作为 SPI 接口进行扩展（1.7.2 版本以前 SlotChainBuilder 作为 SPI），使得 Slot Chain 具备了扩展的能力。可以自行加入自定义的 slot 并编排 slot 间的顺序，从而可以给 Sentinel 添加自定义的功能。

Slot也为两大类：

• 统计数据构建部分（statistic）

• • NodeSelectorSlot
  • ClusterBuilderSlot
  • LogSlot
  • StatisticSlot

• 规则判断部分（rule checking）

• • AuthoritySlot
  • SystemSlot
  • ParamFlowSlot
  • FlowSlot
  • DegradeSlot

Node

Sentinel中的簇点链路是由一个个的Node组成的，Node是一个接口，包括下面的实现：

所有的节点都可以记录对资源的访问统计数据，所以都是StatisticNode的子类。

按照作用分为两类Node：

• DefaultNode：代表链路树中的每一个资源，一个资源出现在不同链路中时，会创建不同的DefaultNode节点。而树的入口节点叫EntranceNode，是一种特殊的DefaultNode
• ClusterNode：代表资源，一个资源不管出现在多少链路中，只会有一个ClusterNode。记录的是当前资源被访问的所有统计数据之和。

DefaultNode记录的是资源在当前链路中的访问数据，用来实现基于链路模式的限流规则。

ClusterNode记录的是资源在所有链路中的访问数据，实现默认模式、关联模式的限流规则。

例如：在一个SpringMVC项目中，有两个业务：

• 业务1：controller中的资源/order/query访问了service中的资源/goods
• 业务2：controller中的资源/order/save访问了service中的资源/goods

创建的链路图如下：

Entry

默认情况下，Sentinel会将controller中的方法作为被保护资源，如果想定义自己的资源，则需显示定义。

Sentinel中的资源用Entry来表示。

定义资源的两种方式

使用try-catch定义资源

// 资源名可使用任意有业务语义的字符串，比如方法名、接口名或其它可唯一标识的字符串。
try (Entry entry = SphU.entry("resourceName")) {// 被保护的业务逻辑// do something here...
} catch (BlockException ex) {// 资源访问阻止，被限流或被降级// 在此处进行相应的处理操作
}

public Order queryOrderById(Long orderId) {// 创建Entry，标记资源，资源名为resource1try (Entry entry = SphU.entry("resource1")) {// 1.查询订单，这里是假数据Order order = Order.build(101L, 4999L, "小米 MIX4", 1, 1L, null);// 2.查询用户，基于Feign的远程调用User user = userClient.findById(order.getUserId());// 3.设置order.setUser(user);// 4.返回return order;}catch (BlockException e){log.error("被限流或降级", e);return null;}
}

使用注解标记资源

@SentinelResource("orderResource")
public Order queryOrderById(Long orderId) {// 1.查询订单Order order = orderMapper.findById(orderId);// 2.用Feign远程调用User user = userClient.findById(order.getUserId());// 3.封装user到Orderorder.setUser(user);// 4.返回return order;
}

基于注解标记资源的实现原理

Sentinel的stater中，spring.factories声明需要就是自动装配的配置类

在SentinelAutoConfiguration类中声明了一个SentinelResourceAspectbean

在SentinelResourceAspect中基于AOP实现了增强。

@Aspect
public class SentinelResourceAspect extends AbstractSentinelAspectSupport {// 切点是添加了 @SentinelResource注解的类@Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")public void sentinelResourceAnnotationPointcut() {}// 环绕增强@Around("sentinelResourceAnnotationPointcut()")public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {// 获取受保护的方法Method originMethod = resolveMethod(pjp);// 获取 @SentinelResource注解// @SentinelResource("resource1")SentinelResource annotation = originMethod.getAnnotation(SentinelResource.class);if (annotation == null) {// Should not go through here.throw new IllegalStateException("Wrong state for SentinelResource annotation");}// 获取注解上的资源名称String resourceName = getResourceName(annotation.value(), originMethod);EntryType entryType = annotation.entryType();int resourceType = annotation.resourceType();Entry entry = null;try {// 创建资源 Entryentry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());// 执行受保护的方法Object result = pjp.proceed();return result;} catch (BlockException ex) {return handleBlockException(pjp, annotation, ex);} catch (Throwable ex) {Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();// The ignore list will be checked first.if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {throw ex;}if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {traceException(ex);return handleFallback(pjp, annotation, ex);}// No fallback function can handle the exception, so throw it out.throw ex;} finally {if (entry != null) {entry.exit(1, pjp.getArgs());}}}
}

@SentinelResource注解就是一个标记，而Sentinel基于AOP思想，对被标记的方法做环绕增强，完成资源（Entry）的创建。

Context

在上面的簇点链路中除了controller方法、service方法两个资源外，还多了一个默认的入口节点：

sentinel_spring_web_context，这是一个EntranceNode类型的节点。

这个节点是在初始化Context的时候由Sentinel创建的。

什么是Context

• Context 代表调用链路上下文，贯穿一次调用链路中的所有资源（ Entry），基于ThreadLocal。
• Context 维持着入口节点（entranceNode）、本次调用链路的 curNode（当前资源节点）、调用来源（origin）等信息。
• 后续的Slot都可以通过Context拿到DefaultNode或者ClusterNode，从而获取统计数据，完成规则判断
• Context初始化的过程中，会创建EntranceNode，contextName就是EntranceNode的名称

对应的API如下：

// 创建context，包含两个参数：context名称、 来源名称
ContextUtil.enter("contextName", "originName");

Context的初始化

在SentinelWebAutoConfiguration中，添加了SentinelWebInterceptor

@Autowired
private Optional<SentinelWebInterceptor> sentinelWebInterceptorOptional;
@Override
public void addInterceptors(InterceptorRegistry registry) {if (!sentinelWebInterceptorOptional.isPresent()) {return;}SentinelProperties.Filter filterConfig = properties.getFilter();// 添加一个SentinelWebInterceptor拦截器registry.addInterceptor(sentinelWebInterceptorOptional.get()).order(filterConfig.getOrder()).addPathPatterns(filterConfig.getUrlPatterns());log.info("[Sentinel Starter] register SentinelWebInterceptor with urlPatterns: {}.",filterConfig.getUrlPatterns());
}

SentinelWebInterceptor继承自AbstractSentinelInterceptor，而AbstractSentinelInterceptor实现了HandlerInterceptor接口，会拦截一切进入controller的方法，执行preHandle前置拦截方法，而Context的初始化就是在这里完成的。

@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
throws Exception {try {// 获取资源名称，一般是controller方法的@RequestMapping路径，例如/order/{orderId}String resourceName = getResourceName(request);if (StringUtil.isEmpty(resourceName)) {return true;}if (increaseReferece(request, this.baseWebMvcConfig.getRequestRefName(), 1) != 1) {return true;}// Parse the request origin using registered origin parser.// 从request中获取请求来源，将来做 授权规则 判断时会用String origin = parseOrigin(request);// 获取 contextName，默认是sentinel_spring_web_contextString contextName = getContextName(request);// 创建 ContextContextUtil.enter(contextName, origin);// 创建资源，名称就是当前请求的controller方法的映射路径Entry entry = SphU.entry(resourceName, ResourceTypeConstants.COMMON_WEB, EntryType.IN);request.setAttribute(baseWebMvcConfig.getRequestAttributeName(), entry);return true;} catch (BlockException e) {try {handleBlockException(request, response, e);} finally {ContextUtil.exit();}return false;}
}

如果要关闭使用sentinel_spring_web_context作为默认的root，只需要在application.yml文件中关闭即可

spring: cloud: sentinel:web-context-unify: false # 关闭context整合

ContextUtil

创建Context的方法就是ContextUtil.enter(contextName, origin);

public static Context enter(String name, String origin) {if (Constants.CONTEXT_DEFAULT_NAME.equals(name)) {throw new ContextNameDefineException("The " + Constants.CONTEXT_DEFAULT_NAME + " can't be permit to defined!");}return trueEnter(name, origin);
}protected static Context trueEnter(String name, String origin) {// 尝试获取contextContext context = contextHolder.get();// 判空if (context == null) {// 如果为空，开始初始化Map<String, DefaultNode> localCacheNameMap = contextNameNodeMap;// 尝试获取入口节点DefaultNode node = localCacheNameMap.get(name);if (node == null) {// 判断缓存数量是否上限if (localCacheNameMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {setNullContext();return NULL_CONTEXT;} else {// 加锁，确保线程安全LOCK.lock();try {node = contextNameNodeMap.get(name);if (node == null) {if (contextNameNodeMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {setNullContext();return NULL_CONTEXT;} else {node = new EntranceNode(new StringResourceWrapper(name, EntryType.IN), null);// 添加入口节点到 ROOTConstants.ROOT.addChild(node);// 将入口节点放入缓存Map<String, DefaultNode> newMap = new HashMap<>(contextNameNodeMap.size() + 1);newMap.putAll(contextNameNodeMap);newMap.put(name, node);contextNameNodeMap = newMap;}}} finally {LOCK.unlock();}}}// 创建Context，参数为：入口节点 和 contextNamecontext = new Context(node, name);// 设置请求来源 origincontext.setOrigin(origin);// 放入ThreadLocalcontextHolder.set(context);}// 返回return context;
}

ProcessorSlotChain执行流程

入口

从上面的分析中可以知道，资源可以分为两种类型

• sentinel自己对所有的controller接口创建的资源

• • 这种资源的创建位于AbstractSentinelInterceptor的preHandle方法中

    

• 我们自己使用注解或者try-cache创建的声明的资源

• • 这种资源的创建位于SentinelResourceAspect的环绕通知中

    

这两种方式都是用了SphU.entry();这个方法

public static Entry entry(String name, int resourceType, EntryType trafficType, Object[] args)
throws BlockException {return Env.sph.entryWithType(name, resourceType, trafficType, 1, args);
}@Override
public Entry entryWithType(String name, int resourceType, EntryType entryType, int count, Object[] args)throws BlockException {return entryWithType(name, resourceType, entryType, count, false, args);
}@Override
public Entry entryWithType(String name, int resourceType, EntryType entryType, int count, boolean prioritized,Object[] args) throws BlockException {// 将 资源名称等基本信息 封装为一个 StringResourceWrapper对象StringResourceWrapper resource = new StringResourceWrapper(name, entryType, resourceType);return entryWithPriority(resource, count, prioritized, args);
}private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args) throws BlockException {// 获取 ContextContext 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);}// 获取 Slot执行链，同一个资源，会创建一个执行链，放入缓存// 这里是创建 DefaultProcessorSlotChain子类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，并将 resource、chain、context 记录在 Entry中Entry e = new CtEntry(resourceWrapper, chain, context);try {// 执行 slotChainchain.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;
}

在这段代码中，会获取ProcessorSlotChain对象，然后基于chain.entry()开始执行slotChain中的每一个Slot。 而这里创建的是其实现类：DefaultProcessorSlotChain。

获取ProcessorSlotChain后会保存到Map中，key是ResourceWrapper，值是ProcessorSlotChain。

所以，一个资源只会有一个ProcessorSlotChain。

此时在DefaultProcessorSlotChain的entry()方法中，first是DefaultProcessorSlotChain，之后基于责任链模式，后续slot作为上一个slot的next执行。

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args) throws Throwable {first.transformEntry(context, resourceWrapper, t, count, prioritized, args);
}

@SuppressWarnings("unchecked")
void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args) throws Throwable {T t = (T)o;entry(context, resourceWrapper, t, count, prioritized, args);
}

AbstractLinkedProcessorSlot中的entry()方法调用顺序如下

NodeSelectorSlot

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
throws Throwable {DefaultNode node = map.get(context.getName());if (node == null) {synchronized (this) {node = map.get(context.getName());if (node == null) {// 创建DefaultNodenode = new DefaultNode(resourceWrapper, null);// 将DefaultNode放入缓存中，key是contextName，// 这样不同链路入口的请求，将会创建多个DefaultNode，相同链路则只有一个DefaultNodeHashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());cacheMap.putAll(map);cacheMap.put(context.getName(), node);map = cacheMap;// Build invocation tree// 将当前资源的DefaultNode设置为上一个资源的childNode((DefaultNode) context.getLastNode()).addChild(node);}}}// 将当前资源的DefaultNode设置为Context中的curNode（当前节点）context.setCurNode(node);fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

这个Slot完成了这么几件事情：

• 为当前资源创建 DefaultNode
• 将DefaultNode放入缓存中，key是contextName，这样不同链路入口的请求，将会创建多个DefaultNode，相同链路则只有一个DefaultNode
• 将当前资源的DefaultNode设置为上一个资源的childNode
• 将当前资源的DefaultNode设置为Context中的curNode（当前节点）

下一个slot，就是ClusterBuilderSlot

ClusterBuilderSlot

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {// 判空，注意ClusterNode是共享的成员变量，也就是说一个资源只有一个ClusterNode，与链路无关if (clusterNode == null) {synchronized (lock) {if (clusterNode == null) {// Create the cluster node.clusterNode = new ClusterNode(resourceWrapper.getName(), resourceWrapper.getResourceType());HashMap<ResourceWrapper, ClusterNode> newMap = new HashMap<>(Math.max(clusterNodeMap.size(), 16));newMap.putAll(clusterNodeMap);// 放入缓存newMap.put(node.getId(), clusterNode);clusterNodeMap = newMap;}}}// 将资源的 DefaultNode与 ClusterNode关联node.setClusterNode(clusterNode);/** if context origin is set, we should get or create a new {@link Node} of the specific origin.*/if (!"".equals(context.getOrigin())) {// 记录请求来源 origin 将 origin放入 entryNode originNode = node.getClusterNode().getOrCreateOriginNode(context.getOrigin());context.getCurEntry().setOriginNode(originNode);}// 继续下一个slotfireEntry(context, resourceWrapper, node, count, prioritized, args);
}

LogSlot

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode obj, int count, boolean prioritized, Object... args)
throws Throwable {try {// 执行下一个slot，如果有异常则记录并抛出fireEntry(context, resourceWrapper, obj, count, prioritized, args);} catch (BlockException e) {EagleEyeLogUtil.log(resourceWrapper.getName(), e.getClass().getSimpleName(), e.getRuleLimitApp(), context.getOrigin(), count);throw e;} catch (Throwable e) {RecordLog.warn("Unexpected entry exception", e);}}

StatisticSlot

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,boolean prioritized, Object... args) throws Throwable {try {// 执行下一个slot，做限流、降级等判断fireEntry(context, resourceWrapper, node, count, prioritized, args);// 请求通过了, 线程计数器 +1 ，用作线程隔离node.increaseThreadNum();// 请求计数器 +1 用作限流node.addPassRequest(count);if (context.getCurEntry().getOriginNode() != null) {// 如果有 origin，来源计数器也都要 +1context.getCurEntry().getOriginNode().increaseThreadNum();context.getCurEntry().getOriginNode().addPassRequest(count);}if (resourceWrapper.getEntryType() == EntryType.IN) {// 如果是入口资源，还要给全局计数器 +1.Constants.ENTRY_NODE.increaseThreadNum();Constants.ENTRY_NODE.addPassRequest(count);}// 请求通过后的回调.for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {handler.onPass(context, resourceWrapper, node, count, args);}} catch (PriorityWaitException ex) {异常处理......throw e;}
}

另外，需要注意的是，所有的计数+1动作都包括两部分，以node.addPassRequest(count);为例：

@Override
public void addPassRequest(int count) {// DefaultNode的计数器，代表当前链路的 计数器super.addPassRequest(count);// ClusterNode计数器，代表当前资源的 总计数器this.clusterNode.addPassRequest(count);
}

AuthoritySlot

负责请求来源origin的授权规则判断，如图：

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)throws Throwable {// 校验黑白名单checkBlackWhiteAuthority(resourceWrapper, context);// 进入下一个 slotfireEntry(context, resourceWrapper, node, count, prioritized, args);
}

void checkBlackWhiteAuthority(ResourceWrapper resource, Context context) throws AuthorityException {// 获取授权规则Map<String, Set<AuthorityRule>> authorityRules = AuthorityRuleManager.getAuthorityRules();if (authorityRules == null) {return;}Set<AuthorityRule> rules = authorityRules.get(resource.getName());if (rules == null) {return;}// 遍历规则并判断for (AuthorityRule rule : rules) {if (!AuthorityRuleChecker.passCheck(rule, context)) {// 规则不通过，直接抛出异常throw new AuthorityException(context.getOrigin(), rule);}}
}

static boolean passCheck(AuthorityRule rule, Context context) {String requester = context.getOrigin();// Empty origin or empty limitApp will pass.if (StringUtil.isEmpty(requester) || StringUtil.isEmpty(rule.getLimitApp())) {return true;}// Do exact match with origin name.int pos = rule.getLimitApp().indexOf(requester);boolean contain = pos > -1;// 如果包含，做精确匹配if (contain) {boolean exactlyMatch = false;// 使用逗号分割String[] appArray = rule.getLimitApp().split(",");for (String app : appArray) {if (requester.equals(app)) {exactlyMatch = true;break;}}contain = exactlyMatch;}// 获取校验方式。0：白名单；1：黑名单int strategy = rule.getStrategy();if (strategy == RuleConstant.AUTHORITY_BLACK && contain) {return false;}if (strategy == RuleConstant.AUTHORITY_WHITE && !contain) {return false;}return true;
}

默认的请求头是空的（""），所以需要使用gateway或者重写RequestOriginParser的parseOrigin()方法对请求头进行处理，都则空的origin会直接通过

重写RequestOriginParser

@Component
public class HeaderOriginParser implements RequestOriginParser {@Overridepublic String parseOrigin(HttpServletRequest request) {// 1.获取请求头String origin = request.getHeader("origin");// 2.非空判断，给空请求头附默认值if (StringUtils.isEmpty(origin)) {origin = "blank";}return origin;}
}

在gateway中对请求头做处理

spring:	cloud:gateway:default-filters:# 给经过网关的请求添加请求头，格式为 xxx,yyy- AddRequestHeader=Truth,Itcast is freaking awesome!- AddRequestHeader=origin,gateway

SystemSlot

SystemSlot是对系统保护的规则校验：

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,boolean prioritized, Object... args) throws Throwable {SystemRuleManager.checkSystem(resourceWrapper);fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

public static void checkSystem(ResourceWrapper resourceWrapper) throws BlockException {if (resourceWrapper == null) {return;}// Ensure the checking switch is on.if (!checkSystemStatus.get()) {return;}// 只针对入口资源做校验，其它直接返回if (resourceWrapper.getEntryType() != EntryType.IN) {return;}// total qpsdouble currentQps = Constants.ENTRY_NODE == null ? 0.0 : Constants.ENTRY_NODE.successQps();if (currentQps > qps) {throw new SystemBlockException(resourceWrapper.getName(), "qps");}// total threadint currentThread = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.curThreadNum();if (currentThread > maxThread) {throw new SystemBlockException(resourceWrapper.getName(), "thread");}// 全局平均 RT校验double rt = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.avgRt();if (rt > maxRt) {throw new SystemBlockException(resourceWrapper.getName(), "rt");}// 全局 系统负载 校验if (highestSystemLoadIsSet && getCurrentSystemAvgLoad() > highestSystemLoad) {if (!checkBbr(currentThread)) {throw new SystemBlockException(resourceWrapper.getName(), "load");}}// 全局 CPU使用率 校验if (highestCpuUsageIsSet && getCurrentCpuUsage() > highestCpuUsage) {throw new SystemBlockException(resourceWrapper.getName(), "cpu");}
}

ParamFlowSlot

是针对进入资源的请求，针对不同的请求参数值分别统计QPS的限流方式。

• 这里的单机阈值，就是最大令牌数量：maxCount
• 这里的统计窗口时长，就是统计时长：duration

含义是每隔duration时间长度内，最多生产maxCount个令牌，上图配置的含义是每1秒钟生产2个令牌。

这里在配置资源名时，默认的controller资源是不生效的，

需要写@SentinelResource("orderResource")中的资源名。

比如对于下面的controller应该写hot而不是/order/{orderId}

@SentinelResource("hot")
@GetMapping("{orderId}")
public Order queryOrderByUserId(@PathVariable("orderId") Long orderId) {// 根据id查询订单并返回return orderService.queryOrderById(orderId);
}

这是因为

// AbstractSentinelInterceptor
// 这里没有传入参数
Entry entry = SphU.entry(resourceName, ResourceTypeConstants.COMMON_WEB, EntryType.IN);// SentinelResourceAspect
// 这里在最后传入的参数pjp.getArgs()
entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());

在dispatchServerlet中的doDispatch方法内部可以看到，先执行applyPreHandle（controller资源），然后执行handle（注解资源）

所以一个请求会创建两次资源（如果这个请求controller方法添加了@SentinelResource("hot")）

protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {HttpServletRequest processedRequest = request;HandlerExecutionChain mappedHandler = null;boolean multipartRequestParsed = false;WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);try {ModelAndView mv = null;Exception dispatchException = null;try {processedRequest = checkMultipart(request);multipartRequestParsed = (processedRequest != request);// Determine handler for the current request.mappedHandler = getHandler(processedRequest);if (mappedHandler == null) {noHandlerFound(processedRequest, response);return;}// Determine handler adapter for the current request.HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());// Process last-modified header, if supported by the handler.String method = request.getMethod();boolean isGet = "GET".equals(method);if (isGet || "HEAD".equals(method)) {long lastModified = ha.getLastModified(request, mappedHandler.getHandler());if (new ServletWebRequest(request, response).checkNotModified(lastModified) && isGet) {return;}}// 前置处理，在这里进行controller接口的资源创建，创建的资源不支持热点参数限流if (!mappedHandler.applyPreHandle(processedRequest, response)) {return;}// 执行真正的处理，在这里对添加注解的方法创建资源，支持热点参数限流// 因此，对于热点参数限流的controller方法，第一次请求进入到ParamFlowSlot的entry// 方法中时，不会进行checkFlow，因为没有传入热点参数// 第二次进入时，才会进行热点参数限流// Actually invoke the handler.mv = ha.handle(processedRequest, response, mappedHandler.getHandler());if (asyncManager.isConcurrentHandlingStarted()) {return;}applyDefaultViewName(processedRequest, mv);mappedHandler.applyPostHandle(processedRequest, response, mv);}catch (Exception ex) {异常处理。。。。。。
}

核心API：

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,boolean prioritized, Object... args) throws Throwable {// 如果没有热点规则，直接放行if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) {fireEntry(context, resourceWrapper, node, count, prioritized, args);return;}// 检查热点规则checkFlow(resourceWrapper, count, args);fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

void checkFlow(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {// 检查args是否为nullif (args == null) {return;}// 检查ParamFlowRuleManager中是否定义了与给定资源名称相关的流量规则if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) {return;}// 从ParamFlowRuleManager中获取与给定资源名称相关的所有流量规则，并将它们存储在rules列表中List<ParamFlowRule> rules = ParamFlowRuleManager.getRulesOfResource(resourceWrapper.getName());// 遍历rules列表中的每个流量规则rulefor (ParamFlowRule rule : rules) {// 应用实际的参数索引paramIdxapplyRealParamIdx(rule, args.length);// 初始化参数指标ParameterMetricStorageParameterMetricStorage.initParamMetricsFor(resourceWrapper, rule);// 使用ParamFlowChecker检查资源是否满足流量规则if (!ParamFlowChecker.passCheck(resourceWrapper, rule, count, args)) {// 获取触发异常的参数值和规则String triggeredParam = "";if (args.length > rule.getParamIdx()) {Object value = args[rule.getParamIdx()];triggeredParam = String.valueOf(value);}throw new ParamFlowException(resourceWrapper.getName(), triggeredParam, rule);}}
}

public static boolean passCheck(ResourceWrapper resourceWrapper, /*@Valid*/ ParamFlowRule rule, /*@Valid*/ int count,Object... args) {if (args == null) {return true;}int paramIdx = rule.getParamIdx();if (args.length <= paramIdx) {return true;}// Get parameter value.Object value = args[paramIdx];// Assign value with the result of paramFlowKey methodif (value instanceof ParamFlowArgument) {value = ((ParamFlowArgument) value).paramFlowKey();}// If value is null, then passif (value == null) {return true;}if (rule.isClusterMode() && rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {return passClusterCheck(resourceWrapper, rule, count, value);}return passLocalCheck(resourceWrapper, rule, count, value);
}private static boolean passLocalCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count,Object value) {try {if (Collection.class.isAssignableFrom(value.getClass())) {for (Object param : ((Collection)value)) {if (!passSingleValueCheck(resourceWrapper, rule, count, param)) {return false;}}} else if (value.getClass().isArray()) {int length = Array.getLength(value);for (int i = 0; i < length; i++) {Object param = Array.get(value, i);if (!passSingleValueCheck(resourceWrapper, rule, count, param)) {return false;}}} else {return passSingleValueCheck(resourceWrapper, rule, count, value);}} catch (Throwable e) {RecordLog.warn("[ParamFlowChecker] Unexpected error", e);}return true;
}static boolean passSingleValueCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int acquireCount,Object value) {if (rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {if (rule.getControlBehavior() == RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER) {return passThrottleLocalCheck(resourceWrapper, rule, acquireCount, value);} else {// 走这里return passDefaultLocalCheck(resourceWrapper, rule, acquireCount, value);}} else if (rule.getGrade() == RuleConstant.FLOW_GRADE_THREAD) {Set<Object> exclusionItems = rule.getParsedHotItems().keySet();long threadCount = getParameterMetric(resourceWrapper).getThreadCount(rule.getParamIdx(), value);if (exclusionItems.contains(value)) {int itemThreshold = rule.getParsedHotItems().get(value);return ++threadCount <= itemThreshold;}long threshold = (long)rule.getCount();return ++threadCount <= threshold;}return true;
}

令牌桶

热点规则判断采用了令牌桶算法来实现参数限流，为每一个不同参数值设置令牌桶，Sentinel的令牌桶有两部分组成：

这两个Map的key都是请求的参数值，value却不同，其中：

• tokenCounters：用来记录剩余令牌数量
• timeCounters：用来记录上一个请求的时间

当一个携带参数的请求到来后，基本判断流程是这样的：

/*** 检查资源是否过载，基于给定的resourceWrapper、rule、acquireCount和value。* * @param resourceWrapper 资源包装器* @param rule 参数流规则* @param acquireCount 要获得的令牌数量* @param value 当前请求的标识符，热点参数* @return 如果资源没有过载，返回true，否则返回false*/
static boolean passDefaultLocalCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int acquireCount, Object value) {// 获取参数指标ParameterMetric metric = getParameterMetric(resourceWrapper);// 获取令牌计数器和时间计数器从指标CacheMap<Object, AtomicLong> tokenCounters = metric == null ? null : metric.getRuleTokenCounter(rule);CacheMap<Object, AtomicLong> timeCounters = metric == null ? null : metric.getRuleTimeCounter(rule);// 检查令牌计数器和时间计数器是否为空if (tokenCounters == null || timeCounters == null) {return true;}// exclusionItems存放单独配置热点限流规则的参数值，对这部分参数做自定义的限流Set<Object> exclusionItems = rule.getParsedHotItems().keySet();long tokenCount = (long)rule.getCount();if (exclusionItems.contains(value)) {tokenCount = rule.getParsedHotItems().get(value);}// 检查令牌数量是否为0if (tokenCount == 0) {return false;}// 计算最大计数（阈值），getBurstCount()允许的突发阈值，一般为0long maxCount = tokenCount + rule.getBurstCount();if (acquireCount > maxCount) {return false;}// Token bucket算法while (true) {// 获取当前时间long currentTime = TimeUtil.currentTimeMillis();// 获取上次添加令牌的时间AtomicLong lastAddTokenTime = timeCounters.putIfAbsent(value, new AtomicLong(currentTime));// 如果上次添加令牌的时间为空，说明令牌从未添加if (lastAddTokenTime == null) {// 令牌未添加，只需补充令牌并消耗 acquireCount 立即返回 truetokenCounters.putIfAbsent(value, new AtomicLong(maxCount - acquireCount));return true;}// 计算自上次添加令牌以来时间间隔long passTime = currentTime - lastAddTokenTime.get();// 一个简化版的 token bucket 算法，当统计窗口已过时，才会补充令牌// 如果当前经过的时间大于一个统计窗口的时长if (passTime > rule.getDurationInSec() * 1000) {AtomicLong oldQps = tokenCounters.putIfAbsent(value, new AtomicLong(maxCount - acquireCount));if (oldQps == null) {// Might not be accurate here.lastAddTokenTime.set(currentTime);return true;} else {long restQps = oldQps.get();// 计算累计令牌数量： 经过时间*每秒生成令牌数/（统计窗口时长*1000）long toAddCount = (passTime * tokenCount) / (rule.getDurationInSec() * 1000);// 去最大令牌数量和累计令牌数量最小值long newQps = toAddCount + restQps > maxCount ? (maxCount - acquireCount) : (restQps + toAddCount - acquireCount);// 如果没有令牌直接返回if (newQps < 0) {return false;}if (oldQps.compareAndSet(restQps, newQps)) {lastAddTokenTime.set(currentTime);return true;}Thread.yield();}} else {// 获取剩余的令牌AtomicLong oldQps = tokenCounters.get(value);// 如果剩余的令牌不为空，比较并设置旧令牌值if (oldQps != null) {long oldQpsValue = oldQps.get();if (oldQpsValue - acquireCount >= 0) {if (oldQps.compareAndSet(oldQpsValue, oldQpsValue - acquireCount)) {return true;}} else {return false;}}// 释放线程Thread.yield();}}
}

FlowSlot

包括：

• 三种流控模式：直接模式、关联模式、链路模式
• 三种流控效果：快速失败、warm up、排队等待

三种流控模式，从底层数据统计角度，分为两类：

• 对进入资源的所有请求（ClusterNode）做限流统计：直接模式、关联模式
• 对进入资源的部分链路（DefaultNode）做限流统计：链路模式

三种流控效果，从限流算法来看，分为两类：

• 滑动时间窗口算法：快速失败、warm up
• 漏桶算法：排队等待效果

@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);
}

public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {if (ruleProvider == null || resource == null) {return;}Collection<FlowRule> rules = ruleProvider.apply(resource.getName());if (rules != null) {for (FlowRule rule : rules) {if (!canPassCheck(rule, context, node, count, prioritized)) {throw new FlowException(rule.getLimitApp(), rule);}}}
}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);
}

private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,boolean prioritized) {// 根绝请求和策略选择nodeNode selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);if (selectedNode == null) {return true;}// 这里的canPass根据不同的策略采用不同的算法实现return rule.getRater().canPass(selectedNode, acquireCount, prioritized);
}

static Node selectNodeByRequesterAndStrategy(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node) {// The limit app should not be empty.String limitApp = rule.getLimitApp();// public static final int STRATEGY_DIRECT = 0;// public static final int STRATEGY_RELATE = 1;// public static final int STRATEGY_CHAIN = 2;int strategy = rule.getStrategy();	String origin = context.getOrigin();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)) {if (strategy == RuleConstant.STRATEGY_DIRECT) {// Return the cluster node.// 直连模式和关联模式都采用clusterNodereturn node.getClusterNode();}return selectReferenceNode(rule, context, node);} else if (RuleConstant.LIMIT_APP_OTHER.equals(limitApp)&& FlowRuleManager.isOtherOrigin(origin, rule.getResource())) {if (strategy == RuleConstant.STRATEGY_DIRECT) {return context.getOriginNode();}return selectReferenceNode(rule, context, node);}return null;
}static Node selectReferenceNode(FlowRule rule, Context context, DefaultNode node) {String refResource = rule.getRefResource();int strategy = rule.getStrategy();if (StringUtil.isEmpty(refResource)) {return null;}// 如果是关联模式，返回clusterNodeif (strategy == RuleConstant.STRATEGY_RELATE) {return ClusterBuilderSlot.getClusterNode(refResource);}// 如果是链路模式，返回当前的node（defaultNode）if (strategy == RuleConstant.STRATEGY_CHAIN) {if (!refResource.equals(context.getName())) {return null;}return node;}// No node.return null;
}

滑动时间窗口

时间窗口请求量统计

这里canPass()对规则的判断先要通过FlowRule#getRater()获取流量控制器TrafficShapingController，然后再做限流

而TrafficShapingController有3种实现：

• DefaultController：快速失败，默认的方式，基于滑动时间窗口算法
• WarmUpController：预热模式，基于滑动时间窗口算法，只不过阈值是动态的
• RateLimiterController：排队等待模式，基于漏桶算法

最终的限流判断都在TrafficShapingController的canPass方法中。

这里进入了DefaultNode内部：

发现同时对DefaultNode和ClusterNode在做QPS统计，DefaultNode和ClusterNode都是StatisticNode的子类，这里调用addPassRequest()方法，最终都会进入StatisticNode中。随便跟入一个：

这里有秒、分两种纬度的统计，对应两个计数器。找到对应的成员变量，可以看到：

两个计数器都是ArrayMetric类型，并且传入了两个参数：

// intervalInMs：是滑动窗口的时间间隔，默认为 1 秒
// sampleCount: 时间窗口的分隔数量，默认为 2，就是把 1秒分为 2个小时间窗
public ArrayMetric(int sampleCount, int intervalInMs) {this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
}

如图：

接下来，我们进入ArrayMetric类的addPass方法：

@Override
public void addPass(int count) {// 获取当前时间所在的时间窗WindowWrap<MetricBucket> wrap = data.currentWindow();// 计数器 +1wrap.value().addPass(count);
}

这里的data是一个LeapArray：

private final LeapArray<MetricBucket> data;

LeapArray的四个属性：

public abstract class LeapArray<T> {// 小窗口的时间长度，默认是500ms ，值 = intervalInMs / sampleCountprotected int windowLengthInMs;// 滑动窗口内的 小窗口 数量，默认为 2protected int sampleCount;// 滑动窗口的时间间隔，默认为 1000msprotected int intervalInMs;// 滑动窗口的时间间隔，单位为秒，默认为 1private double intervalInSecond;
}

LeapArray是一个环形数组，因为时间是无限的，数组长度不可能无限，因此数组中每一个格子放入一个时间窗（window），当数组放满后，角标归0，覆盖最初的window。

因为滑动窗口最多分成sampleCount数量的小窗口，因此数组长度只要大于sampleCount，那么最近的一个滑动窗口内的2个小窗口就永远不会被覆盖，就不用担心旧数据被覆盖的问题了。

data.currentWindow()方法：

public WindowWrap<T> currentWindow(long timeMillis) {if (timeMillis < 0) {return null;}// 计算当前时间对应的数组角标int idx = calculateTimeIdx(timeMillis);// 计算当前时间所在窗口的开始时间.long windowStart = calculateWindowStart(timeMillis);/*** 先根据角标获取数组中保存的 oldWindow 对象，可能是旧数据，需要判断.** (1) oldWindow 不存在, 说明是第一次，创建新 window并存入，然后返回即可* (2) oldWindow的 starTime = 本次请求的 windowStar, 说明正是要找的窗口，直接返回.* (3) oldWindow的 starTime < 本次请求的 windowStar, 说明是旧数据，需要被覆盖，创建 *     新窗口，覆盖旧窗口*/while (true) {WindowWrap<T> old = array.get(idx);if (old == null) {// 创建新 windowWindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));// 基于CAS写入数组，避免线程安全问题if (array.compareAndSet(idx, null, window)) {// 写入成功，返回新的 windowreturn window;} else {// 写入失败，说明有并发更新，等待其它人更新完成即可Thread.yield();}} else if (windowStart == old.windowStart()) {return old;} else if (windowStart > old.windowStart()) {if (updateLock.tryLock()) {try {// 获取并发锁，覆盖旧窗口并返回return resetWindowTo(old, windowStart);} finally {updateLock.unlock();}} else {// 获取锁失败，等待其它线程处理就可以了Thread.yield();}} else if (windowStart < old.windowStart()) {// 这种情况不应该存在，写这里只是以防万一。return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));}}
}

找到当前时间所在窗口（WindowWrap）后，只要调用WindowWrap对象中的add方法，计数器+1即可。

这里只负责统计每个窗口的请求量，不负责拦截。限流拦截要看FlowSlot中的逻辑。

滑动窗口QPS计算

FlowSlot的限流判断最终都由TrafficShapingController接口中的canPass方法来实现。该接口有三个实现类：

• DefaultController：快速失败，默认的方式，基于滑动时间窗口算法
• WarmUpController：预热模式，基于滑动时间窗口算法，只不过阈值是动态的
• RateLimiterController：排队等待模式，基于漏桶算法

跟入默认的DefaultController中的canPass方法来分析：

@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {// 计算目前为止滑动窗口内已经存在的请求量int curCount = avgUsedTokens(node);// 判断：已使用请求量 + 需要的请求量（1） 是否大于 窗口的请求阈值if (curCount + acquireCount > count) {// 大于，说明超出阈值，返回falseif (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;}// 小于等于，说明在阈值范围内，返回truereturn true;
}

因此，判断的关键就是int curCount = avgUsedTokens(node);

private int avgUsedTokens(Node node) {if (node == null) {return DEFAULT_AVG_USED_TOKENS;}return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}

因为采用的是限流，走node.passQps()逻辑：

// 这里又进入了 StatisticNode类
@Override
public double passQps() {// 请求量 ÷ 滑动窗口时间间隔 ，得到的就是QPSreturn rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}

rollingCounterInSecond.pass()得到请求量

// rollingCounterInSecond 本质是ArrayMetric
@Override
public long pass() {// 获取当前窗口data.currentWindow();long pass = 0;// 获取当前时间的滑动窗口范围内的所有小窗口List<MetricBucket> list = data.values();// 遍历for (MetricBucket window : list) {// 累加求和pass += window.pass();}// 返回return pass;
}

data.values()如何获取滑动窗口范围内的所有小窗口：

// 此处进入LeapArray类中：
public List<T> values(long timeMillis) {if (timeMillis < 0) {return new ArrayList<T>();}// 创建空集合，大小等于 LeapArray长度int size = array.length();List<T> result = new ArrayList<T>(size);// 遍历 LeapArrayfor (int i = 0; i < size; i++) {// 获取每一个小窗口WindowWrap<T> windowWrap = array.get(i);// 判断这个小窗口是否在滑动窗口时间范围内（1秒内）if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {// 不在范围内，则跳过continue;}// 在范围内，则添加到集合中result.add(windowWrap.value());}// 返回集合return result;
}

isWindowDeprecated(timeMillis, windowWrap)判断窗口是否符合要求

public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {// 当前时间 - 窗口开始时间  是否大于 滑动窗口的最大间隔（1秒）// 也就是说，我们要统计的时 距离当前时间1秒内的小窗口的 count之和return time - windowWrap.windowStart() > intervalInMs;
}

DegradeSlot

Sentinel的熔断是基于状态机实现的。当达到失败阈值时，断路器会打开，按照配置规则进行熔断。当熔断时间结束后，断路器会进入到 half-open 状态，尝试放行一次请求。当请求成功时断路器会关闭，否则重新回到打开状态。

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,boolean prioritized, Object... args) throws Throwable {performChecking(context, resourceWrapper);fireEntry(context, resourceWrapper, node, count, prioritized, args);
}void performChecking(Context context, ResourceWrapper r) throws BlockException {List<CircuitBreaker> 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());}}
}

@Override
public boolean tryPass(Context context) {// 判断状态机状态if (currentState.get() == State.CLOSED) {// 如果是closed状态，直接放行return true;}if (currentState.get() == State.OPEN) {// 如果是OPEN状态，断路器打开// 继续判断OPEN时间窗是否结束，如果是则把状态从OPEN切换到 HALF_OPEN，返回truereturn retryTimeoutArrived() && fromOpenToHalfOpen(context);}// OPEN状态，并且时间窗未到，返回falsereturn false;
}

protected boolean retryTimeoutArrived() {// 当前时间 大于 下一次 HalfOpen的重试时间return TimeUtil.currentTimeMillis() >= nextRetryTimestamp;
}

protected boolean fromOpenToHalfOpen(Context context) {// 基于CAS修改状态，从 OPEN到 HALF_OPENif (currentState.compareAndSet(State.OPEN, State.HALF_OPEN)) {// 状态变更的事件通知notifyObservers(State.OPEN, State.HALF_OPEN, null);// 得到当前资源Entry entry = context.getCurEntry();// 给资源设置监听器，在资源Entry销毁时（资源业务执行完毕时）触发entry.whenTerminate(new BiConsumer<Context, Entry>() {@Overridepublic void accept(Context context, Entry entry) {// 判断 资源业务是否异常if (entry.getBlockError() != null) {// 如果异常，则再次进入OPEN状态currentState.compareAndSet(State.HALF_OPEN, State.OPEN);notifyObservers(State.HALF_OPEN, State.OPEN, 1.0d);}}});return true;}return false;
}

这里出现了从OPEN到HALF_OPEN、从HALF_OPEN到OPEN的变化，但是还有几个没有：

• 从CLOSED到OPEN
• 从HALF_OPEN到CLOSED

触发断路器

请求经过所有插槽 后，一定会执行exit方法，而在DegradeSlot的exit方法中：

会调用CircuitBreaker的onRequestComplete()方法。而CircuitBreaker有两个实现：

@Override
public void onRequestComplete(Context context) {// 获取资源 EntryEntry entry = context.getCurEntry();if (entry == null) {return;}// 尝试获取 资源中的 异常Throwable error = entry.getError();// 获取计数器，同样采用了滑动窗口来计数SimpleErrorCounter counter = stat.currentWindow().value();if (error != null) {// 如果出现异常，则 error计数器 +1counter.getErrorCount().add(1);}// 不管是否出现异常，total计数器 +1counter.getTotalCount().add(1);// 判断异常比例是否超出阈值handleStateChangeWhenThresholdExceeded(error);
}

阈值判断的方法：

private void handleStateChangeWhenThresholdExceeded(Throwable error) {// 如果当前已经是OPEN状态，不做处理if (currentState.get() == State.OPEN) {return;}// 如果已经是 HALF_OPEN 状态，判断是否需求切换状态if (currentState.get() == State.HALF_OPEN) {if (error == null) {// 没有异常，则从 HALF_OPEN 到 CLOSEDfromHalfOpenToClose();} else {// 有一次，再次进入OPENfromHalfOpenToOpen(1.0d);}return;}// 说明当前是CLOSE状态，需要判断是否触发阈值List<SimpleErrorCounter> 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;}// 按照异常数统计// 如果比例超过阈值，切换到 OPENif (curCount > threshold) {transformToOpen(curCount);}
}