微服务架构 | 5.4 Sentinel 流控、统计和熔断的源码分析(经典)

妖狐艹你老母 2024-03-25 09:54 73阅读 0赞

*  [前言][Link 1]
 *  [1. Sentinel 的自动装配][1. Sentinel]
    
     *  [1.2 依赖引入][1.2]
     *  [1.3 SentinelWebAutoConfiguration 配置类][1.3 SentinelWebAutoConfiguration]
     *  [1.4 CommonFilter 过滤器][1.4 CommonFilter]
     *  [1.5 小结][1.5]
 *  [2. 获取 ProcessorSlot 链][2. _ ProcessorSlot]
    
     *  [2.1 Sentinel 源码包结构][2.1 Sentinel]
     *  [2.2 获取 ProcessorSlot 链与操作 Slot 槽的入口 CtSph.entryWithPriority()][2.2 _ ProcessorSlot _ Slot _ CtSph.entryWithPriority]
        
         *  [2.2.1 构造 ProcessorSlot 链 CtSph.lookProcessChain()][2.2.1 _ ProcessorSlot _ CtSph.lookProcessChain]
         *  [2.2.2 操作 Slot 槽的入口][2.2.2 _ Slot]
 *  [3. 流控槽实施流控逻辑 FlowSlot.entry()][3. _ FlowSlot.entry]
    
     *  [3.1 获取流控规则 FlowSlot.ruleProvider.apply()][3.1 _ FlowSlot.ruleProvider.apply]
     *  [3.2 校验每条规则 FlowRuleChecker.canPassCheck()][3.2 _ FlowRuleChecker.canPassCheck]
        
         *  [3.2.1 获取 Node FlowRuleChecker.selectNodeByRequesterAndStrategy()][3.2.1 _ Node FlowRuleChecker.selectNodeByRequesterAndStrategy]
         *  [3.2.2 获取流控的处理策略 \`FlowRule.getRater().canPass()][3.2.2 _ _FlowRule.getRater_.canPass]
 *  [4. 统计槽实施指标数据统计 StatisticSlot.entry()][4. _ StatisticSlot.entry]
    
     *  [4.1 统计“增加线程数”和“请求通过数”][4.1]
     *  [4.2 数据统计的数据结构][4.2]
        
         *  [4.2.1 ArrayMetric 指标数组][4.2.1 ArrayMetric]
         *  [4.2.2 LeapArray 环形数组][4.2.2 LeapArray]
         *  [4.2.3 WindowWrap 窗口包装类][4.2.3 WindowWrap]
         *  [4.2.4 MetricBucket 指标桶][4.2.4 MetricBucket]
         *  [4.2.5 各数据结构的依赖关系][4.2.5]
         *  [4.2.6 LeapArray 统计数据的大致思路][4.2.6 LeapArray]
 *  [5. 熔断槽实施服务熔断 DegradeSlot.entry()][5. _ DegradeSlot.entry]
    
     *  [5.1 继续或取消熔断功能][5.1]
     *  [5.2 请求失败，启动熔断][5.2]
 *  [6. Sentinel 源码结构图小结][6. Sentinel]
 *  [最后][Link 2]

--------------------

## **前言** ##

**参考资料**：  
《Spring Microservices in Action》  
《Spring Cloud Alibaba 微服务原理与实战》  
《B站 尚硅谷 SpringCloud 框架开发教程 周阳》  
[《Sentinel GitHub 官网》][Sentinel GitHub]  
[《Sentinel 官网》][Sentinel]

调用链路是 Sentinel 的工作主流程，由各个 Slot 槽组成，将不同的 Slot 槽按照顺序串在一起，从而将不同的功能（限流、降级、系统保护）组合在一起；

本篇《2. 获取 ProcessorSlot 链》将从源码级讲解如何获取调用链路，接着会以遍历链表的方式处理每一个 Slot 槽，其中就有：FlowSlot、StatisticSlot、DegradeSlot 等。分别对应本篇《3. 流控槽实施流控逻辑》、《4. 统计槽实施指标数据统计》和《5. 熔断槽实施服务熔断》；

--------------------

## **1. Sentinel 的自动装配** ##

### **1.2 依赖引入** ###

*  我们引入 Sentinel 的 starter 依赖文件，不需要太多额外操作，即可使用 Sentinel 默认自带的限流功能，原因是这些配置和功能都给我们自动装配了；
 *  在 Spring-Cloud-Alibaba-Sentinel 包下的 META-INF/spring.factories 文件里定义了会自动装配哪些类；

![Sentinel 的自动装配][Sentinel 1]

*  **SentinelWebAutoConfiguration**：对 Web Servlet 环境的支持；
 *  **SentinelWebFluxAutoConfiguration**：对 Spring WebFlux 的支持；
 *  **SentinelEndpointAutoConfiguration**：暴露 Endpoint 信息；
 *  **SentinelFeignAutoConfiguration**：用于适应 Feign 组件；
 *  **SentinelAutoConfiguration**：支持对 RestTemplate 的服务调用使用 Sentinel 进行保护；

### **1.3 SentinelWebAutoConfiguration 配置类** ###

*  在 **SentinelWebAutoConfiguration** 配置类中自动装配了一个 **FilterRegistrationBean**，其主要作用是注册一个 CommonFilter，并且默认情况下通过 `/*` 规则拦截所有的请求；

<code class="language-java">@Configuration
 @EnableConfigurationProperties(SentinelProperties.class)
 public class SentinelWebAutoConfiguration {
 
 //省略其他代码
 
 	@Bean
 	@ConditionalOnProperty(name = "spring.cloud.sentinel.filter.enabled", matchIfMissing = true)
 	public FilterRegistrationBean sentinelFilter() {
 		FilterRegistrationBean<Filter> registration = new FilterRegistrationBean<>();
 
 		SentinelProperties.Filter filterConfig = properties.getFilter();
 
 		if (filterConfig.getUrlPatterns() == null || filterConfig.getUrlPatterns().isEmpty()) {
 			List<String> defaultPatterns = new ArrayList<>();
 			//默认情况下通过 /* 规则拦截所有的请求
 			defaultPatterns.add("/*");
 			filterConfig.setUrlPatterns(defaultPatterns);
 		}
 
 		registration.addUrlPatterns(filterConfig.getUrlPatterns().toArray(new String[0]));
 		//【点进去】注册 CommonFilter
 		Filter filter = new CommonFilter();
 		registration.setFilter(filter);
 		registration.setOrder(filterConfig.getOrder());
 		registration.addInitParameter("HTTP_METHOD_SPECIFY", String.valueOf(properties.getHttpMethodSpecify()));
 		log.info("[Sentinel Starter] register Sentinel CommonFilter with urlPatterns: {}.", filterConfig.getUrlPatterns());
 		return registration;
 	}
 }
 </code>

### **1.4 CommonFilter 过滤器** ###

*  **CommonFilter** 过滤器的作用与源码如下：
    
     *  从请求中获取目标 URL；
     *  获取 Urlcleaner；
     *  对当前 URL 添加限流埋点；

### **1.5 小结** ###

*  对于 Web Servlet 环境，只是通过 Filter 的方式将所有请求自动设置为 Sentinel 的资源，从而达到限流的目的；

## **2. 获取 ProcessorSlot 链** ##

*  Sentinel 的工作原理主要依靠 **ProcessorSlot** 链，遍历链中的每一个 **Slot** 槽，执行相应逻辑；

### **2.1 Sentinel 源码包结构** ###

*  在 DeBug 之前，我们需要对 Sentinel 的源码包结构做个分析，以找到方法的入口；

<table> 
 <thead> 
 <tr> 
 <th style="background-color:#fafafa;">模块名</th> 
 <th style="background-color:#fafafa;">说明</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td style="border-color:#c0c0c0;">sentinel-adapter</td> 
 <td style="border-color:#c0c0c0;">负责针对主流开源框架进行限流适配，如：Dubbo、gRPC、Zuul 等；</td> 
 </tr> 
 <tr> 
 <td style="border-color:#c0c0c0;">sentinel-core</td> 
 <td style="border-color:#c0c0c0;">Sentinel 核心库，提供限流、熔断等实现；</td> 
 </tr> 
 <tr> 
 <td style="border-color:#c0c0c0;">sentinel-dashboard</td> 
 <td style="border-color:#c0c0c0;">控制台模块，提供可视化监控和管理；</td> 
 </tr> 
 <tr> 
 <td style="border-color:#c0c0c0;">sentinel-demo</td> 
 <td style="border-color:#c0c0c0;">官方案例；</td> 
 </tr> 
 <tr> 
 <td style="border-color:#c0c0c0;">sentinel-extension</td> 
 <td style="border-color:#c0c0c0;">实现不同组件的数据源扩展，如：Nacos、ZooKeeper、Apollo 等；</td> 
 </tr> 
 <tr> 
 <td style="border-color:#c0c0c0;">sentinel-transport</td> 
 <td style="border-color:#c0c0c0;">通信协议处理模块；</td> 
 </tr> 
 </tbody> 
</table>

*  Slot 槽是 Sentinel 的核心，因此方法的入口在 sentinel-core 核心库，里面有好多个 `SphU.entry()` 方法，我们给方法打上断点，DeBug 进入，然后登录 Sentinel 控制台；

![首次DeBug 进入 SphU.entry() 方法][DeBug _ SphU.entry_]

### **2.2 获取 ProcessorSlot 链与操作 Slot 槽的入口 CtSph.entryWithPriority()** ###

*  一直进入最终方法的实现在 `CtSph.entryWithPriority()` 方法里，其主要逻辑与源码如下：
    
     *  校验全局上下文 context；
     *  构造 ProcessorSlot 链；
     *  遍历 ProcessorSlot 链操作 Slot 槽（遍历链表）；

<code class="language-java">private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args) throws BlockException {
 Context context = ContextUtil.getContext();
 if (context instanceof NullContext) {
 //上下文量已经超过阈值 -> 只初始化条目，不进行规则检查
 return new CtEntry(resourceWrapper, null, context);
 }
 
 if (context == null) {
 //没有指定上下文 -> 使用默认上下文 context
 context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
 }
 
 if (!Constants.ON) {
 //全局开关关闭 -> 没有规则检查
 return new CtEntry(resourceWrapper, null, context);
 }
 //【断点步入 2.2.1】通过 lookProcessChain 方法获取 ProcessorSlot 链
 ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
 
 
 if (chain == null) {
 //表示资源量超过 Constants.MAX_SLOT_CHAIN_SIZE 常量 -> 不会进行规则检查
 return new CtEntry(resourceWrapper, null, context);
 }
 
 Entry e = new CtEntry(resourceWrapper, chain, context);
 try {
 //【断点步入 3./4./5.】执行 ProcessorSlot 对 ProcessorSlot 链中的 Slot 槽遍历操作（遍历链表的方式）
 chain.entry(context, resourceWrapper, null, count, prioritized, args);
 } catch (BlockException e1) {
 e.exit(count, args);
 throw e1;
 } catch (Throwable e1) {
 //这种情况不应该发生，除非 Sentinel 内部存在错误
 RecordLog.info("Sentinel unexpected exception", e1);
 }
 return e;
 }
 </code>

#### **2.2.1 构造 ProcessorSlot 链 CtSph.lookProcessChain()** ####

*  进入 `CtSph.lookProcessChain()` 方法；

<code class="language-java">ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
 //从缓存中获取 slot 调用链
 ProcessorSlotChain chain = chainMap.get(resourceWrapper);
 if (chain == null) {
 synchronized (LOCK) {
 chain = chainMap.get(resourceWrapper);
 if (chain == null) {
 // Entry size limit.
 if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
 return null;
 }
 //【断点步入】构造 Slot 链（责任链模式）
 chain = SlotChainProvider.newSlotChain();
 Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
 chainMap.size() + 1);
 newMap.putAll(chainMap);
 newMap.put(resourceWrapper, chain);
 chainMap = newMap;
 }
 }
 }
 return chain;
 }
 </code>

*  最终调用 `DefaultSlotChainBuilder.build()` 方法构造 DefaultProcessorSlotChain；

<code class="language-java">@Override
 public ProcessorSlotChain build() {
 ProcessorSlotChain chain = new DefaultProcessorSlotChain();
 List<ProcessorSlot> 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;
 }
 </code>

*  可以看到最后 ProcessorSlotChain 链中有 10 个 Slot 插槽：
 *  在本篇笔记中我们关注 3 个槽：
    
     *  **FlowSlot**：进行流控规则校验，对应本篇《3. 流控槽实施流控逻辑》；
     *  **StatisticSlot**：实现指标数据的统计，对应本篇《4. 统计槽实施指标数据统计》；
     *  **DegradeSlot**：服务熔断，对应本篇《5. 熔断槽实施服务熔断》

![ProcessorSlotChain 链中有 10 个 Slot 插槽][ProcessorSlotChain _ 10 _ Slot]

#### **2.2.2 操作 Slot 槽的入口** ####

*  操作 Slot 槽的入口方法是：`ProcessorSlot.entry()`；
 *  接着会以遍历链表的方式操作每个 Slot 槽，其中就有：FlowSlot、StatisticSlot、DegradeSlot 等。分别对应下面的《3. 流控槽实施流控逻辑》、《4. 统计槽实施指标数据统计》和《5. 熔断槽实施服务熔断》；

## **3. 流控槽实施流控逻辑 FlowSlot.entry()** ##

*  进入 `ProcessorSlot.entry()` 方法，它会遍历每个 Slot 插槽，并对其进行操作，其中会经过 `FlowSlot.entry()` 方法（需要提前给该方法打上断点），方法的逻辑跟源码如下：

*  进入 `FlowSlot.checkFlow()` 方法，最终调用 `FlowRuleChecker.checkFlow()` 方法，方法的逻辑和源码如下：
    
     *  遍历所有流控规则 FlowRule；
     *  针对每个规则调用 canPassCheck 进行校验；

<code class="language-java">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;
 }
 //【断点步入 3.1】获取流控规则
 Collection<FlowRule> 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);
 }
 }
 }
 }
 </code>

### **3.1 获取流控规则 FlowSlot.ruleProvider.apply()** ###

*  进入 `FlowSlot.ruleProvider.apply()` 方法，获取到 Sentinel 控制台上的流控规则；

<code class="language-java">private final Function<String, Collection<FlowRule>> ruleProvider = new Function<String, Collection<FlowRule>>() {
 @Override
 public Collection<FlowRule> apply(String resource) {
 // Flow rule map should not be null.
 Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();
 return flowRules.get(resource);
 }
 };
 </code>

### **3.2 校验每条规则 FlowRuleChecker.canPassCheck()** ###

*  进入 `FlowRuleChecker.canPassCheck()` 方法，分集群和单机模式校验每条规则；

<code class="language-java">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);
 }
 </code>

*  由于我们是单机模式，进入 `FlowRuleChecker.passLocalCheck()` 方法，其主要逻辑和源码如下：
    
     *  根据来源和策略获取 Node，从而拿到统计的 runtime 信息；
     *  使用流量控制器检查是否让流量通过；

<code class="language-java">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);
 }
 </code>

#### **3.2.1 获取 Node FlowRuleChecker.selectNodeByRequesterAndStrategy()** ####

*  进入 `FlowRuleChecker.selectNodeByRequesterAndStrategy()` 方法，其根据 FlowRule 中配置的 Strategy 和 limitApp 属性，返回不同处理策略的 Node；

<code class="language-java">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;
 }
 </code>

*  假设我们对接口 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**，这是一个接口，有四种实现类，如下图所示：

![TrafficShapingController 的四种实现类][TrafficShapingController]

*  有以下四种处理策略：
    
     *  **DefaultController**：直接拒绝；
     *  **RateLimiterController**：匀速排队；
     *  **WarmUpController**：冷启动（预热）；
     *  **WarmUpRateLimiterController**：匀速+冷启动。
 *  最终调用 `TrafficShapingController.canPass()` 方法，执行流控行为；

## **4. 统计槽实施指标数据统计 StatisticSlot.entry()** ##

*  限流的核心是限流算法的实现，Sentinel 默认采用滑动窗口算法来实现限流，具体的指标数据统计由 StatisticSlot 实现；
 *  我们给 `StatisticSlot.entry()` 方法里的语句打上断点，运行到光标处；
 *  `StatisticSlot.entry()` 方法的核心是使用 Node 统计“增加线程数”和“请求通过数”；

<code class="language-java">@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<DefaultNode> 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<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
 handler.onPass(context, resourceWrapper, node, count, args);
 }
 //处理限流、熔断等异常 
 } catch (BlockException e) {
 
 //省略
 
 throw e;
 //处理业务异常 
 } catch (Throwable e) {
 context.getCurEntry().setError(e);
 throw e;
 }
 }
 </code>

### **4.1 统计“增加线程数”和“请求通过数”** ###

*  这两个方法都是调用同一个类的，笔者以第一个为例，进入 `DefaultNode.increaseThreadNum()` 方法，最终调用的是 `StatisticNode.increaseThreadNum()`，而统计也是依靠 **StatisticNode** 维护的，这里放上 **StatisticNode** 的统计核心与源码：
    
     *  StatisticNode 持有两个计数器 Metric 对象，统计行为是通过 Metric 完成的；

*  这里还有减少请求通过数（线程数）、统计最大值等方法，由于篇幅有限，这里不放出，感兴趣的读者可以自己 DeBug 进入看看；

### **4.2 数据统计的数据结构** ###

#### **4.2.1 ArrayMetric 指标数组** ####

*  **ArrayMetric** 的构造方法需要先给方法打上断点，重新 DeBug，在初始化时注入构造；

#### **4.2.2 LeapArray 环形数组** ####

*  **LeapArray** 是处理数据的核心数据结构，采用滑动窗口算法；
 *  **ArrayMetric** 中持有 **LeapArray** 对象，所有方法都是对 **LeapArray** 进行操作；
 *  **LeapArray** 是环形的数据结构，为了节约内存，它存储固定个数的窗口对象 WindowWrap，只保存最近一段时间的数据，新增的时间窗口会覆盖最早的时间窗口；

<code class="language-java">public abstract class LeapArray<T> {
 
 //省略其他代码
 
 //单个窗口的长度（1个窗口多长时间）
 protected int windowLengthInMs;
 //采样窗口个数
 protected int sampleCount;
 //全部窗口的长度（全部窗口多长时间）
 protected int intervalInMs;
 private double intervalInSecond;
 //窗口数组：存储所有窗口（支持原子读取和写入）
 protected final AtomicReferenceArray<WindowWrap<T>> 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<T> currentWindow() {
 //这里参数是当前时间
 return currentWindow(TimeUtil.currentTimeMillis());
 }
 //获取指定时间的窗口
 public WindowWrap<T> 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<T> old = array.get(idx);
 //（1）窗口不在
 if (old == null) {
 //创建一个窗口
 WindowWrap<T> window = new WindowWrap<T>(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<T>(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;
 }
 }
 </code>

#### **4.2.3 WindowWrap 窗口包装类** ####

*  **WindowWrap** 是一个窗口对象，它是一个包装类，包装的对象是 **MetricBucket**；

<code class="language-java">public class WindowWrap<T> {
 //窗口长度，与 LeapArray 的 windowLengthInMs 一致
 private final long windowLengthInMs;
 //窗口开始时间，其值是 windowLengthInMs 的整数倍
 private long windowStart;
 //窗口的数据，支持 MetricBucket 类型，存储统计数据
 private T value;
 
 //省略其他代码
 }
 </code>

#### **4.2.4 MetricBucket 指标桶** ####

*  **MetricBucket** 类的定义如下，可以发现指标数据存在 **LongAdder**\[\] counters中；
 *  **LongAdder** 是 JDK1.8 中新增的类，用于在高并发场景下代替AtomicLong，以用空间换时间的方式降低了 CAS 失败的概率，从而提高性能；

<code class="language-java">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);
 }
 
 //省略其他代码
 }
 </code>

#### **4.2.5 各数据结构的依赖关系** ####

![各数据结构的 UML 图.png][UML _.png]

![结构示意图.png][.png]

#### **4.2.6 LeapArray 统计数据的大致思路** ####

*  创建一个长度为 n 的数组，数组元素就是窗口，窗口包装了 1 个指标桶，桶中存放了该窗口时间范围中对应的请求统计数据；
 *  可以想象成一个环形数组在时间轴上向右滚动，请求到达时，会命中数组中的一个窗口，那么该请求的数据就会存到命中的这个窗口包含的指标桶中；
 *  当数组转满一圈时，会回到数组的开头，而此时下标为 0 的元素需要重复使用，它里面的窗口数据过期了，需要重置，然后再使用。具体过程如下图：

![LeapArray 统计数据的大致思路][LeapArray]

## **5. 熔断槽实施服务熔断 DegradeSlot.entry()** ##

*  服务熔断是通过 **DegradeSlot** 来实现的，它会根据用户配置的熔断规则和系统运行时各个 Node 中的统计数据进行熔断判断；
 *  注意：熔断功能在 Sentinel-1.8.0 版本前后有较大变化；
 *  我们给 `DegradeSlot.entry()` 方法里的语句打上断点，运行到光标处；

*  进入 `DegradeSlot.performChecking()` 方法，其逻辑与源码如下：
    
     *  根据资源名称获取断路器；
     *  循环判断每个断路器；

<code class="language-java">void performChecking(Context context, ResourceWrapper r) throws BlockException {
 //根据 resourceName 获取断路器
 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());
 }
 }
 }
 </code>

### **5.1 继续或取消熔断功能** ###

*  进入 `AbstractCircuitBreaker.tryPass()` 方法，当请求超时并且处于探测恢复（半开状态，HALF-OPEN 状态）失败时继续断路功能；

*  进入 `AbstractCircuitBreaker.fromOpenToHalfOpen()` 方法，实现状态的变更；

<code class="language-java">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<Context, Entry>() {
 @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;
 }
 </code>

*  上述讲解了：状态从 OPEN 变为 HALF\_OPEN，HALF\_OPEN 变为 OPEN；
 *  但状态从 HALF\_OPEN 变为 CLOSE 需要在正常执行完请求后，由 entry.exit() 调用 `DegradeSlot.exit()` 方法来改变状态；

### **5.2 请求失败，启动熔断** ###

*  状态从 HALF\_OPEN 变为 CLOSE 的实现方法在 `DegradeSlot.exit()`；

*  进入 `ExceptionCircuitBreaker.onRequestComplete()` 方法，其主要逻辑与源码如下：
    
     *  请求失败比例与总请求比例加 1，用于判断后续是否超过阈值；

*  进入 `ExceptionCircuitBreaker.handleStateChangeWhenThresholdExceeded()` 方法，变更状态；

<code class="language-java">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<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;
 }
 if (curCount > threshold) {
 transformToOpen(curCount);
 }
 }
 </code>

## **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**：环形数组；
                                
                                 *  **WindowWrap**：窗口包装类；
                             *  **MetricBucket**：指标桶；
                 *  **DefaultNode.addPassRequest()**：统计“增加线程数”；
                    
                     *  **StatisticNode.addPassRequest()**：同上；
             *  **DegradeSlot.entry()**：遍历到 DegradeSlot 槽，服务熔断；
                
                 *  **DegradeSlot.performChecking()**：执行检查；
                    
                     *  **DegradeRuleManager.getCircuitBreakers()**：根据 resourceName 获取断路器；
                     *  **AbstractCircuitBreaker.tryPass()**：继续或取消熔断功能；
                        
                         *  **AbstractCircuitBreaker.fromOpenToHalfOpen()**：尝试通过半开状态；
             *  **DegradeSlot.exit()**：请求失败（超时），启动熔断；
                
                 *  **ExceptionCircuitBreaker.onRequestComplete()**：在请求完成时操作；
                    
                     *  **ExceptionCircuitBreaker.handleStateChangeWhenThresholdExceeded()**：变更状态；

--------------------

## **最后** ##

新人制作，如有错误，欢迎指出，感激不尽！欢迎关注公众号，会分享一些更日常的东西！

如需转载，请标注出处！

来源：[https://www.cnblogs.com/dlhjw/p/15858198.html][https_www.cnblogs.com_dlhjw_p_15858198.html]

[Link 1]: https://www.cnblogs.com/dlhjw/p/15858198.html#%E5%89%8D%E8%A8%80
[1. Sentinel]: https://www.cnblogs.com/dlhjw/p/15858198.html#1-sentinel-%E7%9A%84%E8%87%AA%E5%8A%A8%E8%A3%85%E9%85%8D
[1.2]: https://www.cnblogs.com/dlhjw/p/15858198.html#12-%E4%BE%9D%E8%B5%96%E5%BC%95%E5%85%A5
[1.3 SentinelWebAutoConfiguration]: https://www.cnblogs.com/dlhjw/p/15858198.html#13-sentinelwebautoconfiguration-%E9%85%8D%E7%BD%AE%E7%B1%BB
[1.4 CommonFilter]: https://www.cnblogs.com/dlhjw/p/15858198.html#14-commonfilter-%E8%BF%87%E6%BB%A4%E5%99%A8
[1.5]: https://www.cnblogs.com/dlhjw/p/15858198.html#15-%E5%B0%8F%E7%BB%93
[2. _ ProcessorSlot]: https://www.cnblogs.com/dlhjw/p/15858198.html#2-%E8%8E%B7%E5%8F%96-processorslot-%E9%93%BE
[2.1 Sentinel]: https://www.cnblogs.com/dlhjw/p/15858198.html#21-sentinel-%E6%BA%90%E7%A0%81%E5%8C%85%E7%BB%93%E6%9E%84
[2.2 _ ProcessorSlot _ Slot _ CtSph.entryWithPriority]: https://www.cnblogs.com/dlhjw/p/15858198.html#22-%E8%8E%B7%E5%8F%96-processorslot-%E9%93%BE%E4%B8%8E%E6%93%8D%E4%BD%9C-slot-%E6%A7%BD%E7%9A%84%E5%85%A5%E5%8F%A3-ctsphentrywithpriority
[2.2.1 _ ProcessorSlot _ CtSph.lookProcessChain]: https://www.cnblogs.com/dlhjw/p/15858198.html#221-%E6%9E%84%E9%80%A0-processorslot-%E9%93%BE--ctsphlookprocesschain
[2.2.2 _ Slot]: https://www.cnblogs.com/dlhjw/p/15858198.html#222-%E6%93%8D%E4%BD%9C-slot-%E6%A7%BD%E7%9A%84%E5%85%A5%E5%8F%A3
[3. _ FlowSlot.entry]: https://www.cnblogs.com/dlhjw/p/15858198.html#3-%E6%B5%81%E6%8E%A7%E6%A7%BD%E5%AE%9E%E6%96%BD%E6%B5%81%E6%8E%A7%E9%80%BB%E8%BE%91-flowslotentry
[3.1 _ FlowSlot.ruleProvider.apply]: https://www.cnblogs.com/dlhjw/p/15858198.html#31-%E8%8E%B7%E5%8F%96%E6%B5%81%E6%8E%A7%E8%A7%84%E5%88%99-flowslotruleproviderapply
[3.2 _ FlowRuleChecker.canPassCheck]: https://www.cnblogs.com/dlhjw/p/15858198.html#32-%E6%A0%A1%E9%AA%8C%E6%AF%8F%E6%9D%A1%E8%A7%84%E5%88%99-flowrulecheckercanpasscheck
[3.2.1 _ Node FlowRuleChecker.selectNodeByRequesterAndStrategy]: https://www.cnblogs.com/dlhjw/p/15858198.html#321-%E8%8E%B7%E5%8F%96-node-flowrulecheckerselectnodebyrequesterandstrategy
[3.2.2 _ _FlowRule.getRater_.canPass]: https://www.cnblogs.com/dlhjw/p/15858198.html#322-%E8%8E%B7%E5%8F%96%E6%B5%81%E6%8E%A7%E7%9A%84%E5%A4%84%E7%90%86%E7%AD%96%E7%95%A5-flowrulegetratercanpass
[4. _ StatisticSlot.entry]: https://www.cnblogs.com/dlhjw/p/15858198.html#4-%E7%BB%9F%E8%AE%A1%E6%A7%BD%E5%AE%9E%E6%96%BD%E6%8C%87%E6%A0%87%E6%95%B0%E6%8D%AE%E7%BB%9F%E8%AE%A1-statisticslotentry
[4.1]: https://www.cnblogs.com/dlhjw/p/15858198.html#41-%E7%BB%9F%E8%AE%A1%E5%A2%9E%E5%8A%A0%E7%BA%BF%E7%A8%8B%E6%95%B0%E5%92%8C%E8%AF%B7%E6%B1%82%E9%80%9A%E8%BF%87%E6%95%B0
[4.2]: https://www.cnblogs.com/dlhjw/p/15858198.html#42-%E6%95%B0%E6%8D%AE%E7%BB%9F%E8%AE%A1%E7%9A%84%E6%95%B0%E6%8D%AE%E7%BB%93%E6%9E%84
[4.2.1 ArrayMetric]: https://www.cnblogs.com/dlhjw/p/15858198.html#421-arraymetric-%E6%8C%87%E6%A0%87%E6%95%B0%E7%BB%84
[4.2.2 LeapArray]: https://www.cnblogs.com/dlhjw/p/15858198.html#422-leaparray-%E7%8E%AF%E5%BD%A2%E6%95%B0%E7%BB%84
[4.2.3 WindowWrap]: https://www.cnblogs.com/dlhjw/p/15858198.html#423-windowwrap-%E7%AA%97%E5%8F%A3%E5%8C%85%E8%A3%85%E7%B1%BB
[4.2.4 MetricBucket]: https://www.cnblogs.com/dlhjw/p/15858198.html#424-metricbucket-%E6%8C%87%E6%A0%87%E6%A1%B6
[4.2.5]: https://www.cnblogs.com/dlhjw/p/15858198.html#425-%E5%90%84%E6%95%B0%E6%8D%AE%E7%BB%93%E6%9E%84%E7%9A%84%E4%BE%9D%E8%B5%96%E5%85%B3%E7%B3%BB
[4.2.6 LeapArray]: https://www.cnblogs.com/dlhjw/p/15858198.html#426-leaparray-%E7%BB%9F%E8%AE%A1%E6%95%B0%E6%8D%AE%E7%9A%84%E5%A4%A7%E8%87%B4%E6%80%9D%E8%B7%AF
[5. _ DegradeSlot.entry]: https://www.cnblogs.com/dlhjw/p/15858198.html#5-%E7%86%94%E6%96%AD%E6%A7%BD%E5%AE%9E%E6%96%BD%E6%9C%8D%E5%8A%A1%E7%86%94%E6%96%AD-degradeslotentry
[5.1]: https://www.cnblogs.com/dlhjw/p/15858198.html#51-%E7%BB%A7%E7%BB%AD%E6%88%96%E5%8F%96%E6%B6%88%E7%86%94%E6%96%AD%E5%8A%9F%E8%83%BD
[5.2]: https://www.cnblogs.com/dlhjw/p/15858198.html#52-%E8%AF%B7%E6%B1%82%E5%A4%B1%E8%B4%A5%E5%90%AF%E5%8A%A8%E7%86%94%E6%96%AD
[6. Sentinel]: https://www.cnblogs.com/dlhjw/p/15858198.html#6-sentinel-%E6%BA%90%E7%A0%81%E7%BB%93%E6%9E%84%E5%9B%BE%E5%B0%8F%E7%BB%93
[Link 2]: https://www.cnblogs.com/dlhjw/p/15858198.html#%E6%9C%80%E5%90%8E
[Sentinel GitHub]: https://github.com/alibaba/Sentinel
[Sentinel]: https://sentinelguard.io/zh-cn/
[Sentinel 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/f133049b0d49470f9e9a48f88d195dd5.png
[DeBug _ SphU.entry_]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/861eb3dc7c3c45e6a5ecff43ce20ee3b.png
[ProcessorSlotChain _ 10 _ Slot]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/1a8391e5b00b4a8bbe2ccf8f84ebd8dd.png
[TrafficShapingController]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/23802a469e444b489b4fe22b47560392.png
[UML _.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/ed1afa2beec64c00883bcaaad691e4fe.png
[.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/30ac25863451463ba1c6b17c673279e7.png
[LeapArray]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/dac27c5640e147e7b74fb98497e3a7f7.png
[https_www.cnblogs.com_dlhjw_p_15858198.html]: https://www.cnblogs.com/dlhjw/p/15858198.html