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

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

前言

参考资料：
《Spring Microservices in Action》
《Spring Cloud Alibaba 微服务原理与实战》
《B站 尚硅谷 SpringCloud 框架开发教程 周阳》
《Sentinel GitHub 官网》
《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/域名ories 文件里定义了会自动装配哪些类；

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

1.3 SentinelWebAutoConfiguration 配置类

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

@Configuration
@EnableConfigurationProperties(域名s)
public class SentinelWebAutoConfiguration {
    
    //省略其他代码
    
	@Bean
	@ConditionalOnProperty(name = "域名域名led", matchIfMissing = true)
	public FilterRegistrationBean sentinelFilter() {
		FilterRegistrationBean<Filter> registration = new FilterRegistrationBean<>();

		域名er filterConfig = 域名ilter();

		if (域名rlPatterns() == null || 域名rlPatterns().isEmpty()) {
			List<String> defaultPatterns = new ArrayList<>();
			//默认情况下通过 /* 规则拦截所有的请求
			域名("/*");
			域名rlPatterns(defaultPatterns);
		}

		域名rlPatterns(域名rlPatterns().toArray(new String[0]));
		//【点进去】注册 CommonFilter
		Filter filter = new CommonFilter();
		域名ilter(filter);
		域名rder(域名rder());
		域名nitParameter("HTTP_METHOD_SPECIFY", 域名eOf(域名ttpMethodSpecify()));
		域名("[Sentinel Starter] register Sentinel CommonFilter with urlPatterns: {}.", 域名rlPatterns());
		return registration;
	}
}

1.4 CommonFilter 过滤器

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

public class CommonFilter implements Filter {
    
    //省略部分代码

    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
        HttpServletRequest sRequest = (HttpServletRequest)request;
        Entry urlEntry = null;
        try {
            //解析请求 URL
            String target = 域名erTarget(sRequest);
            //URL 清洗
            UrlCleaner urlCleaner = 域名rlCleaner();
            if (urlCleaner != null) {
                //如果存在，则说明配置过 URL 清洗策略，替换配置的 targer
                target = 域名n(target);
            }
            if (!域名pty(target)) {
                String origin = 域名eOrigin(sRequest);
                域名r("sentinel_web_servlet_context", origin);
                if (域名MethodSpecify) {
                    String pathWithHttpMethod = 域名ethod().toUpperCase() + ":" + target;
                    //使用 域名y() 方法对 URL 添加限流埋点
                    urlEntry = 域名y(pathWithHttpMethod, 1, 域名);
                } else {
                    urlEntry = 域名y(target, 1, 域名);
                }
            }
            //执行过滤
            域名lter(request, response);
        } catch (BlockException var14) {
            HttpServletResponse sResponse = (HttpServletResponse)response;
            域名rlBlockHandler().blocked(sRequest, sResponse, var14);
        } catch (ServletException | RuntimeException | IOException var15) {
            域名eEntry(var15, urlEntry);
            throw var15;
        } finally {
            if (urlEntry != null) {
                域名();
            }
            域名();
        }
    }
}

1.5 小结

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

2. 获取 ProcessorSlot 链

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

2.1 Sentinel 源码包结构

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

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

2.2 获取 ProcessorSlot 链与操作 Slot 槽的入口 域名yWithPriority()

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

private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args) throws BlockException {
    Context context = 域名ontext();
    if (context instanceof NullContext) {
        //上下文量已经超过阈值 -> 只初始化条目，不进行规则检查
        return new CtEntry(resourceWrapper, null, context);
    }

    if (context == null) {
        //没有指定上下文 -> 使用默认上下文 context
        context = 域名rnalEnter(域名EXT_DEFAULT_NAME);
    }
     
     if (!域名) {
        //全局开关关闭 -> 没有规则检查
        return new CtEntry(resourceWrapper, null, context);
    }
    //【断点步入 2.2.1】通过 lookProcessChain 方法获取 ProcessorSlot 链
    ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);


    if (chain == null) {
        //表示资源量超过 域名SLOT_CHAIN_SIZE 常量 -> 不会进行规则检查
        return new CtEntry(resourceWrapper, null, context);
    }

    Entry e = new CtEntry(resourceWrapper, chain, context);
    try {
        //【断点步入 3./4./5.】执行 ProcessorSlot 对 ProcessorSlot 链中的 Slot 槽遍历操作（遍历链表的方式）
        域名y(context, resourceWrapper, null, count, prioritized, args);
    } catch (BlockException e1) {
        域名(count, args);
        throw e1;
    } catch (Throwable e1) {
        //这种情况不应该发生，除非 Sentinel 内部存在错误
        域名("Sentinel unexpected exception", e1);
    }
    return e;
}

2.2.1 构造 ProcessorSlot 链 域名ProcessChain()

• 进入 域名ProcessChain() 方法；

ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
    //从缓存中获取 slot 调用链
    ProcessorSlotChain chain = 域名(resourceWrapper);
    if (chain == null) {
        synchronized (LOCK) {
            chain = 域名(resourceWrapper);
            if (chain == null) {
                // Entry size limit.
                if (域名() >= 域名SLOT_CHAIN_SIZE) {
                    return null;
                }
                //【断点步入】构造 Slot 链（责任链模式）
                chain = 域名lotChain();
                Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
                    域名() + 1);
                域名ll(chainMap);
                域名(resourceWrapper, chain);
                chainMap = newMap;
            }
        }
    }
    return chain;
}

• 最终调用 域名d() 方法构造 DefaultProcessorSlotChain；

@Override
public ProcessorSlotChain build() {
    ProcessorSlotChain chain = new DefaultProcessorSlotChain();
    List<ProcessorSlot> sortedSlotList = 域名(域名s).loadInstanceListSorted();
    for (ProcessorSlot slot : sortedSlotList) {
        if (!(slot instanceof AbstractLinkedProcessorSlot)) {
            域名("The ProcessorSlot(" + 域名lass().getCanonicalName() + ") is not an instance of AbstractLinkedProcessorSlot, can\'t be added into ProcessorSlotChain");
            continue;
        }
        域名ast((AbstractLinkedProcessorSlot<?>) slot);
    }
    return chain;
}

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

2.2.2 操作 Slot 槽的入口

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

3. 流控槽实施流控逻辑 域名y()

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

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

• 进入 域名kFlow() 方法，最终调用 域名kFlow() 方法，方法的逻辑和源码如下：
  • 遍历所有流控规则 FlowRule；
  • 针对每个规则调用 canPassCheck 进行校验；

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 = 域名y(域名ame());
    if (rules != null) {
        //遍历所有流控规则 FlowRule
        for (FlowRule rule : rules) {
            //【点进去 3.2】校验每条规则
            if (!canPassCheck(rule, context, node, count, prioritized)) {
                throw new FlowException(域名imitApp(), rule);
            }
        }
    }
}

3.1 获取流控规则 域名y()

• 进入 域名y() 方法，获取到 Sentinel 控制台上的流控规则；

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 = 域名lowRuleMap();
        return 域名(resource);
    }
};

3.2 校验每条规则 域名assCheck()

• 进入 域名assCheck() 方法，分集群和单机模式校验每条规则；

public boolean canPassCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node, int acquireCount, boolean prioritized) {
    String limitApp = 域名imitApp();
    if (limitApp == null) {
        return true;
    }
    //集群模式
    if (域名usterMode()) {
        return passClusterCheck(rule, context, node, acquireCount, prioritized);
    }
    //【点进去】单机模式
    return passLocalCheck(rule, context, node, acquireCount, prioritized);
}

• 由于我们是单机模式，进入 域名LocalCheck() 方法，其主要逻辑和源码如下：
  • 根据来源和策略获取 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 域名ater().canPass(selectedNode, acquireCount, prioritized);
}

3.2.1 获取 Node 域名ctNodeByRequesterAndStrategy()

• 进入 域名ctNodeByRequesterAndStrategy() 方法，其根据 FlowRule 中配置的 Strategy 和 limitApp 属性，返回不同处理策略的 Node；

static Node selectNodeByRequesterAndStrategy(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node) {
    //limitApp 不能为空
    String limitApp = 域名imitApp();
    int strategy = 域名trategy();
    String origin = 域名rigin();
    
    //场景1：限流规则设置了具体应用，如果当前流量就是通过该应用的，则命中场景1
    if (域名ls(origin) && filterOrigin(origin)) {
        if (strategy == 域名TEGY_DIRECT) {
            // Matches limit origin, return origin statistic node.
            return 域名riginNode();
        }
        return selectReferenceNode(rule, context, node);
    } else if (域名ls(limitApp)) {
    //场景2：限流规则未指定任何具体应，默认为default，则当前流量直接命中场景2
        if (strategy == 域名TEGY_DIRECT) {
            // Return the cluster node.
            return 域名lusterNode();
        }

        return selectReferenceNode(rule, context, node);
    } else if (域名ls(limitApp) && 域名herOrigin(origin, 域名esource())) {
    //场景3：限流规则设置的是other，当前流量未命中前两种场景
        if (strategy == 域名TEGY_DIRECT) {
            return 域名riginNode();
        }
        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 获取流控的处理策略 `域名ater().canPass()

• 进入 域名ater().canPass() 方法，首先通过 域名ater() 获得流控行为 TrafficShapingController，这是一个接口，有四种实现类，如下图所示：

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

4. 统计槽实施指标数据统计 域名y()

• 限流的核心是限流算法的实现，Sentinel 默认采用滑动窗口算法来实现限流，具体的指标数据统计由 StatisticSlot 实现；
• 我们给 域名y() 方法里的语句打上断点，运行到光标处；
• 域名y() 方法的核心是使用 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 统计“增加线程数”和“请求通过数”
        域名easeThreadNum();
        域名assRequest(count);

        //如果存在来源节点，则对来源节点增加线程数和请求通过数
        if (域名urEntry().getOriginNode() != null) {
            域名urEntry().getOriginNode().increaseThreadNum();
            域名urEntry().getOriginNode().addPassRequest(count);
        }
        
        //如果是入口流量，则对全局节点增加线程数和请求通过数
        if (域名ntryType() == 域名) {
            域名easeThreadNum();
            域名assRequest(count);
        }

        //执行事件通知和回调函数
        for (ProcessorSlotEntryCallback<DefaultNode> handler : 域名ntryCallbacks()) {
            域名ss(context, resourceWrapper, node, count, args);
        }
    //处理优先级等待异常    
    } catch (PriorityWaitException ex) {
        域名easeThreadNum();
        //如果有来源节点，则对来源节点增加线程数
        if (域名urEntry().getOriginNode() != null) {
            域名urEntry().getOriginNode().increaseThreadNum();
        }

        //如果是入口流量，对全局节点增加线程数
        if (域名ntryType() == 域名) {
            域名easeThreadNum();
        }
        //执行事件通知和回调函数
        for (ProcessorSlotEntryCallback<DefaultNode> handler : 域名ntryCallbacks()) {
            域名ss(context, resourceWrapper, node, count, args);
        }
    //处理限流、熔断等异常    
    } catch (BlockException e) {
        
        //省略
        
        throw e;
    //处理业务异常    
    } catch (Throwable e) {
        域名urEntry().setError(e);
        throw e;
    }
}

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

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

public class StatisticNode implements Node {

    //省略其他代码

    //【断点步入】最近 1s 滑动窗口计数器（默认 1s）
    private transient volatile Metric rollingCounterInSecond = new ArrayMetric(域名LE_COUNT,
        域名RVAL);

    //最近 1min 滑动窗口计数器（默认 1min）
    private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
    
    //增加 “请求通过数” 
    @Override
    public void addPassRequest(int count) {
        域名ass(count);
        域名ass(count);
    }
    //增加 RT 和成功数
    @Override
    public void addRtAndSuccess(long rt, int successCount) {
        域名uccess(successCount);
        域名T(rt);
        域名uccess(successCount);
        域名T(rt);
    }

    //增加“线程数”
    @Override
    public void increaseThreadNum() {
        域名ement();
    }
}

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

4.2 数据统计的数据结构

4.2.1 ArrayMetric 指标数组

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

public class ArrayMetric implements Metric {
    
    //省略其他代码

    //【点进去 4.2.2】数据存储
    private final LeapArray<MetricBucket> data;
    
    //最近 1s 滑动计数器用的是 OccupiableBucketLeapArray
    public ArrayMetric(int sampleCount, int intervalInMs) {
        域名 = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
    }
    
    //最近 1min 滑动计数器用的是 BucketLeapArray
    public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy) {
        if (enableOccupy) {
            域名 = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
        } else {
            域名 = new BucketLeapArray(sampleCount, intervalInMs);
        }
    }

    //增加成功数
    @Override
    public void addSuccess(int count) {
        WindowWrap<MetricBucket> wrap = 域名entWindow();
        域名e().addSuccess(count);
    }

    //增加通过数
    @Override
    public void addPass(int count) {
        WindowWrap<MetricBucket> wrap = 域名entWindow();
        域名e().addPass(count);
    }

    //增加 RT
    @Override
    public void addRT(long rt) {
        WindowWrap<MetricBucket> wrap = 域名entWindow();
        域名e().addRT(rt);
    }
}

4.2.2 LeapArray 环形数组

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

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) {
        //计算单个窗口的长度
        域名owLengthInMs = intervalInMs / sampleCount;
        域名rvalInMs = intervalInMs;
        域名rvalInSecond = intervalInMs / 1000.0;
        域名leCount = sampleCount;
        域名y = new AtomicReferenceArray<>(sampleCount);
    }
    //【点进去 4.2.3】获取当前窗口
    public WindowWrap<T> currentWindow() {
        //这里参数是当前时间
        return currentWindow(域名entTimeMillis());
    }
    //获取指定时间的窗口
    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 = 域名(idx);
            //（1）窗口不在
            if (old == null) {
                //创建一个窗口
                WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
                //CAS将窗口写进 array 中并返回（CAS 操作确保只初始化一次）
                if (域名areAndSet(idx, null, window)) {
                    return window;
                } else {
                    //并发写失败，释放 CPU 资源，避免有线程长时间占用 CPU，一般下次来的时候 array 中有数据了会命中第2种情况；
                    域名d();
                }
            //（2）array 中窗口开始时间 = 当前窗口开始时间
            } else if (windowStart == 域名owStart()) {
                //直接返回
                return old;
            //（3）array 中窗口开始时间 < 当前窗口开始时间    
            } else if (windowStart > 域名owStart()) {
                //尝试获取更新锁
                if (域名ock()) {
                    try {
                        //拿到锁的线程才重置窗口
                        return resetWindowTo(old, windowStart);
                    } finally {
                        //释放锁
                        域名ck();
                    }
                } else {
                    //并发加锁失败，释放 CPU 资源，避免有线程长时间占用 CPU，一般下次来的时候因为 old 对象时间更新了会命中第 2 种情况；
                    域名d();
                }
            //理论上不会出现    
            } else if (windowStart < 域名owStart()) {
                // 正常情况不会进入该分支（机器时钟回拨等异常情况）
                return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
            }
        }
    }
    //计算索引
    private int calculateTimeIdx(/*@Valid*/ long timeMillis) {
        //timeId 降低时间精度
        long timeId = timeMillis / windowLengthInMs;
        //计算当前索引，这样我们就可以将时间戳映射到 leap 数组
        return (int)(timeId % 域名th());
    }
    //计算窗口开始时间
    protected long calculateWindowStart(/*@Valid*/ long timeMillis) {
        return timeMillis - timeMillis % windowLengthInMs;
    }
}

4.2.3 WindowWrap 窗口包装类

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

public class WindowWrap<T> {
    //窗口长度，与 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 = 域名es();
        域名ters = new LongAdder[域名th];
        for (MetricEvent event : events) {
            counters[域名nal()] = new LongAdder();
        }
        initMinRt();
    }

    //覆盖指标
    public MetricBucket reset(MetricBucket bucket) {
        for (MetricEvent event : 域名es()) {
            counters[域名nal()].reset();
            counters[域名nal()].add(域名(event));
        }
        initMinRt();
        return this;
    }

    private void initMinRt() {
        域名t = 域名isticMaxRt();
    }

    //重置指标为0
    public MetricBucket reset() {
        for (MetricEvent event : 域名es()) {
            counters[域名nal()].reset();
        }
        initMinRt();
        return this;
    }
    //获取指标，从 counters 中返回
    public long get(MetricEvent event) {
        return counters[域名nal()].sum();
    }
    //添加指标
    public MetricBucket add(MetricEvent event, long n) {
        counters[域名nal()].add(n);
        return this;
    }

    public long pass() {
        return get(域名);
    }

    public long block() {
        return get(域名K);
    }

    public void addPass(int n) {
        add(域名, n);
    }

    public void addBlock(int n) {
        add(域名K, n);
    }

    //省略其他代码
}

4.2.5 各数据结构的依赖关系

4.2.6 LeapArray 统计数据的大致思路

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

5. 熔断槽实施服务熔断 域名y()

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

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

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

void performChecking(Context context, ResourceWrapper r) throws BlockException {
    //根据 resourceName 获取断路器
    List<CircuitBreaker> circuitBreakers = 域名ircuitBreakers(域名ame());
    if (circuitBreakers == null || 域名pty()) {
        return;
    }
    //循环判断每个断路器
    for (CircuitBreaker cb : circuitBreakers) {
        //【点进去】尝试通过断路器
        if (!域名ass(context)) {
            throw new DegradeException(域名ule().getLimitApp(), 域名ule());
        }
    }
}

5.1 继续或取消熔断功能

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

@Override
public boolean tryPass(Context context) {
    //当前断路器状态为关闭
    if (域名() == 域名ED) {
        return true;
    }
    if (域名() == 域名) {
        //【点进去】对于半开状态，我们尝试通过
        return retryTimeoutArrived() && fromOpenToHalfOpen(context);
    }
    return false;
}

• 进入 域名OpenToHalfOpen() 方法，实现状态的变更；

protected boolean fromOpenToHalfOpen(Context context) {
    //尝试将状态从 OPEN 设置为 HALF_OPEN
    if (域名areAndSet(域名, 域名_OPEN)) {
        //状态变化通知
        notifyObservers(域名, 域名_OPEN, null);
        Entry entry = 域名urEntry();
        //在 entry 添加一个 exitHandler  域名() 时会调用
        域名Terminate(new BiConsumer<Context, Entry>() {
            @Override
            public void accept(Context context, Entry entry) {
                //如果有发生异常，重新将状态设置为OPEN 请求不同通过
                if (域名lockError() != null) {
                    域名areAndSet(域名_OPEN, 域名);
                    notifyObservers(域名_OPEN, 域名, 域名);
                }
            }
        });
        //此时状态已设置为HALF_OPEN正常通行
        return true;
    }
    //熔断
    return false;
}

• 上述讲解了：状态从 OPEN 变为 HALF_OPEN，HALF_OPEN 变为 OPEN；
• 但状态从 HALF_OPEN 变为 CLOSE 需要在正常执行完请求后，由 域名() 调用 域名() 方法来改变状态；

5.2 请求失败，启动熔断

• 状态从 HALF_OPEN 变为 CLOSE 的实现方法在 域名()；

@Override
public void exit(Context context, ResourceWrapper r, int count, Object... args) {
    Entry curEntry = 域名urEntry();
    //无阻塞异常
    if (域名lockError() != null) {
        fireExit(context, r, count, args);
        return;
    }
    //通过资源名获取断路器
    List<CircuitBreaker> circuitBreakers = 域名ircuitBreakers(域名ame());
    //没有配置断路器，则直接放行
    if (circuitBreakers == null || 域名pty()) {
        fireExit(context, r, count, args);
        return;
    }

    if (域名lockError() == null) {
        for (CircuitBreaker circuitBreaker : circuitBreakers) {
            //【点进去】在请求完成时
            域名questComplete(context);
        }
    }
    fireExit(context, r, count, args);
}

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

@Override
public void onRequestComplete(Context context) {
    Entry entry = 域名urEntry();
    if (entry == null) {
        return;
    }
    Throwable error = 域名rror();
    //简单错误计数器
    SimpleErrorCounter counter = 域名entWindow().value();
    if (error != null) {
        //异常请求数加 1
        域名rrorCount().add(1);
    }
    //总请求数加 1
    域名otalCount().add(1);
    //【点进去】超过阈值时变更状态
    handleStateChangeWhenThresholdExceeded(error);
}

• 进入 域名leStateChangeWhenThresholdExceeded() 方法，变更状态；

private void handleStateChangeWhenThresholdExceeded(Throwable error) {
    //全开则直接放行
    if (域名() == 域名) {
        return;
    }
    //半开状态
    if (域名() == 域名_OPEN) {
        //检查请求
        if (error == null) {
            //发生异常，将状态从半开 HALF_OPEN 转为关闭 CLOSE
            fromHalfOpenToClose();
        } else {
            //无异常，解开半开状态
            fromHalfOpenToOpen(域名);
        }
        return;
    }
    
    //计算是否超过阈值
    List<SimpleErrorCounter> counters = 域名es();
    long errCount = 0;
    long totalCount = 0;
    for (SimpleErrorCounter counter : counters) {
        errCount += 域名();
        totalCount += 域名();
    }
    if (totalCount < minRequestAmount) {
        return;
    }
    double curCount = errCount;
    if (strategy == DEGRADE_GRADE_EXCEPTION_RATIO) {
        //熔断策略为：异常比例
        curCount = errCount * 域名 / totalCount;
    }
    if (curCount > threshold) {
        transformToOpen(curCount);
    }
}

6. Sentinel 源码结构图小结

• 域名y()：核心逻辑的入口函数；
  • 域名yWithPriority()：获取 Slot 链，操作 Slot 槽；
    • 域名ProcessChain()：获取 ProcessorSlot 链；
      • 域名d()：构造 DefaultProcessorSlotChain 链（里面有 10 个 Slot 插槽）；
    • 域名y()：遍历 ProcessorSlot 链；

      • 域名y()：遍历到 FlowSlot 槽，限流规则；

        • 域名kFlow()：检查流量规则；
          • 域名kFlow()：使用检查器检查流量规则；
            • 域名y()：获取流控规则；
            • 域名assCheck()：校验每条规则；
              • 域名ClusterCheck()：集群模式；
              • 域名LocalCheck()：单机模式；
                • 域名ctNodeByRequesterAndStrategy()：获取 Node；
                • 域名ater()：获得流控行为 TrafficShapingController；
                • 域名ass()：执行流控行为；

      • 域名y：遍历到 StatisticSlot 槽，统计数据；

        • 域名easeThreadNum()：统计“增加线程数”；
          • 域名easeThreadNum()：统计“请求通过数”；
            • 域名yMetric()：初始化指标数组；
              • LeapArray：环形数组；
                • WindowWrap：窗口包装类；
              • MetricBucket：指标桶；
        • 域名assRequest()：统计“增加线程数”；
          • 域名assRequest()：同上；

      • 域名y()：遍历到 DegradeSlot 槽，服务熔断；

        • 域名ormChecking()：执行检查；
          • 域名ircuitBreakers()：根据 resourceName 获取断路器；
          • 域名ass()：继续或取消熔断功能；
            • 域名OpenToHalfOpen()：尝试通过半开状态；

      • 域名()：请求失败（超时），启动熔断；

        • 域名questComplete()：在请求完成时操作；
          • 域名leStateChangeWhenThresholdExceeded()：变更状态；

最后

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