查看原文
其他

Sentinel FlowSlot 限流实现原理(文末附流程图与总结)

丁威 中间件兴趣圈 2022-11-10

点击上方“中间件兴趣圈”选择“设为星标”

做积极的人,越努力越幸运!

Sentinel 触发限流的实现类为 FlowSlot。我们再来简单思考一下,要实现触发限流,至少需要完成如下几件事情:

  • 收集实时调用信息。

  • 设置触发限流规则

  • 根据限流规则与调用信息来决定是否对请求进行限流等。

如何收集实时调用信息在前面的文章中已详细介绍,请带着上述问题开始本节的探讨。

1、初始 FlowSlot


我们先从 FlotSlot 类的注释来简单认识一下流量控制相关的内容。

  • 根据已(NodeSelectorSlot、ClusterNodeBuilderSlot 和 StatisticSlot)收集的运行时统计信息,FlowSlot将使用预先设置的规则来决定是否应阻止传入请求。

  • SphU.entry(resourceName)调用时,如果有任意一条规则被触发则会抛出 FlowException 异常,应用程序可捕捉该异常对业务进行定制化处理。

  • 每一条流控规则(FlowRule)都包含三个要素:流控类别、基于调用链的流控制策略、限流后的处理行为(参考FlowRule相关的注释)。

    • grade 流量控制的阈值类型
      可选值:QPS(基于QPS限流策略)、并发线程数。

    • strategy 基于调用链的流控制策略
      可选值:STRATEGY_DIRECT(根据调用方限流策略)、STRATEGY_RELATE(关联流量限
      流策略)、STRATEGY_CHAIN(根据调用链入口限流策略)

    • controlBehavior 流量控制后的采取的行为
      CONTROL_BEHAVIOR_DEFAULT(直接拒绝)、CONTROL_BEHAVIOR_WARM_UP(预热)、CONTROL_BEHAVIOR_RATE_LIMITER(匀速排队)、
      CONTROL_BEHAVIOR_WARM_UP_RATE_LIMITER(预热与匀速排队)。

2、FlowSlot 详解


2.1 FlowSlot 类图

FlowSlot 的类图非常简单,内部持有一个成员变量,FlowRuleChecker,用来判断是否满足流控触发条件。

在继续探讨 Sentinel 限流之前,我们先来了解一下 FlowRule,即认识一下 Sentienl 流控规则主要包含哪些配置项,为后续的流程做一个消息的准备。

2.2 FlowRule 配置项

FlowRule 的类体系如图所示:

其属性的含义如下:
  • String resource
    资源的名称。

  • String limitApp
    需要限制的调用来源,对应【新增流控规则界面】的针对来源。

  • int grade
    流量控制的阈值类型,目前支持 QPS 与 并发线程数,对应 【新增流控规则界面】的阔值类型。

  • int strategy
    基于调用链的流量控制策略,对应【新增流控规则界面】的流控模式,其可选取值在本文开头部分有详细介绍。

  • String refResource
    关联资源或入口资源,当流控模式为关联或链路时配置的关联资源或入口资源,对应【新增流控规则界面】的【入口资源】

  • int controlBehavior
    流量控制后的采取的行为,其可选取值在本文开头部分有详细介绍,对应【新增流控规则界面】的流控效果。

  • int warmUpPeriodSec
    预热时间,如果 controlBehavior 设置为预热(warm up)时,可以配置其预热时间,在【新增流控规则界面】中选择 warm up 类型后,会增加一行,供用户配置,默认值 10s。

  • int maxQueueingTimeMs
    最大超时时间,如果 controlBehavior 设置为排队等待时,等待的最大超时时间,默认为500ms。

  • boolean clusterMode
    是否是集群限流模式,对应【新增流控规则界面】的是否集群。

  • ClusterFlowConfig clusterConfig
    集群扩容相关配置,集群限流将在后续文章中重点介绍。

在 sentinel-dashboard 的配置界面如下图所示:

2.3 FlowSlot#entry 流程详解

FlowSlot#entry

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable 
{   // @1
    checkFlow(resourceWrapper, context, node, count, prioritized);       // @2
    fireEntry(context, resourceWrapper, node, count, prioritized, args);  // @3
}

代码@1:首先来解释一下该方法的参数:

  • Context context
    当前 Sentinel 调用的上下文。

  • ResourceWrapper resourceWrapper
    当前访问的资源。

  • DefaultNode node
    当前上下文环境对应的节点。

  • int count
    本次调用需要消耗的“令牌”个数

  • boolean prioritized
    是否是高优先级。

  • Object… args
    额外参数。

代码@2:调用 checkFlow ,根据配置的限流规则,结合实时统计信息,判断是否满足流控条件,如果满足,则触发流控,稍后会详细探讨该方法的实现原理。

代码@3:调用 fireEntry 继续沿着 slot 链进行传播。

FlowSlot 的 checkFlow 方法在内部就是直接调用 FlowRuleChecker 的 checkFlow 方法,故我们将目光放到 FlowRuleChecker 中。

2.4 FlowRuleChecker checkFlow 方法详解

FlowRuleChecker#checkFlow

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());   // @1
    if (rules != null) {
        for (FlowRule rule : rules) {
            if (!canPassCheck(rule, context, node, count, prioritized)) {            // @2
                throw new FlowException(rule.getLimitApp(), rule);
            }
        }
    }
}

代码@1:通过限流规则提供器获取与该资源相关的流控规则列表。

代码@2:然后遍历流控规则列表,通过调用 canPassCheck 方法来判断是否满足该规则设置的条件,如果满足流控规则,则抛出 FlowException,即只需要满足一个即结束校验。

接下来继续查看 canPassCheck 方法。

2.4.1 FlowRuleChecker canPassCheck 详解
public boolean canPassCheck(FlowRule rule, Context context, DefaultNode node, 
                int acquireCount, boolean prioritized) 
{
    String limitApp = rule.getLimitApp(); 
    if (limitApp == null) {    // @1
        return true;
    }
    if (rule.isClusterMode()) {  // @2
        return passClusterCheck(rule, context, node, acquireCount, prioritized);  
    }
    return passLocalCheck(rule, context, node, acquireCount, prioritized);     
}

代码@1:如果限流规则没有配置针对来源,则直接默认通过,该值在配置时,默认为 default,即对所有调用发起方都生效。

代码@2:如果是集群限流模式,则调用 passClusterCheck,非集群限流模式则调用 passLocalCheck 方法,本文重点讲述单节点限流,集群限流模式将在后续文章中详细探讨。

FlowRuleChecker#passLocalCheck

private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
        boolean prioritized) 
{
    Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);    // @1
    if (selectedNode == null) {
        return true;
    }
    return rule.getRater().canPass(selectedNode, acquireCount, prioritized);    // @2
}

代码@1:首先根据流控模式(strategy)选择一个合适的 Node,看到这,大家可以思考一下,这一步骤的目的,如果为空,则直接返回 true,表示放行。

代码@2:调用 FlowRule 内部持有的流量控制器来判断是否符合流控规则,最终调用的是 TrafficShapingController canPass 方法。

那我们接下来分别对上述两个方法进行详细展开。

2.4.1.1 selectNodeByRequesterAndStrategy
FlowRuleChecker#selectNodeByRequesterAndStrategy
static Node selectNodeByRequesterAndStrategy(FlowRule rule, Context context, DefaultNode node) 
{
    String limitApp = rule.getLimitApp();
    int strategy = rule.getStrategy();
    String origin = context.getOrigin();   // @1
    if (limitApp.equals(origin) && filterOrigin(origin)) {    // @2
        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)) {  // @3
        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())) {    // @4
        if (strategy == RuleConstant.STRATEGY_DIRECT) {
            return context.getOriginNode();
        }
        return selectReferenceNode(rule, context, node);
    }
    return null;
}

在介绍该方法之前,先回答上文提到一个问题,我们知道,要判断是否满足了限流规则所配置的条件,一个重要的点就是要拿到当前的实时统计信息,通过上面介绍限流规则时提到 Sentinel 目前支持3种流控模式(直接、关联、链路),针对模式的不同,选择的实时统计数据的逻辑就应该不同,即该方法主要是根据流控策略找到对应的实时统计信息(Node)。

代码@1:首先先介绍几个局部变量的含义:

  • String limitApp
    该条限流规则针对的调用方。

  • int strategy
    该条限流规则的流控策略。

  • String origin
    本次请求的调用方,从当前上下文环境中获取,例如 dubbo 服务提供者,原始调用方为 dubbo 服务提供者的 application。

代码@2:如果限流规则配置的针对的调用方与当前请求实际调用来源匹配(并且不是 default、other)时的处理逻辑,其实现的要点:

  • 如果流控模式为 RuleConstant.STRATEGY_DIRECT(直接),则从 context 中获取源调用方所代表的 Node。

  • 如果流控模式为 RuleConstant.STRATEGY_RELATE(关联),则从集群环境中获取对应关联资源所代表的 Node,通过(ClusterBuilderSlot会收集每一个资源的实时统计信息,子集群限流时详细介绍)

  • 如果流控模式为 RuleConstant.STRATEGY_CHAIN(调用链),则判断当前调用上下文的入口资源与规则配置的是否一样,如果是,则返回入口资源对应的 Node,否则返回 null,注意:返回空则该条流控规则直接通过。【这部分代码,对应代码中的 selectReferenceNode 方法】

代码@3:如果流控规则针对的调用方(limitApp) 配置的为 default,表示对所有的调用源都生效,其获取实时统计节点(Node)的处理逻辑为:

  • 如果流控模式为 RuleConstant.STRATEGY_DIRECT,则直接获取本次调用上下文环境对应的节点的ClusterNode。

  • 如果是其他流控模式,与代码@2的获取逻辑一样,都是调用 selectReferenceNode 进行获取。

代码@4:如果流控规则针对的调用方为(other),此时需要判断是否有针对当前的流控规则,只要存在,则这条规则对当前资源“失效”,如果针对该资源没有配置其他额外的流控规则,则获取实时统计节点(Node)的处理逻辑为:

  • 如果流控模式为 RuleConstant.STRATEGY_DIRECT(直接),则从 context 中获取源调用方所代表的 Node。

  • 如果是其他流控模式,与代码@2的获取逻辑一样,都是调用 selectReferenceNode 进行获取。

从这里可以看出,流控规则针对调用方如果设置为 other,表示针对没有配置流控规则的资源。

根据流控策略选择合适的 Node 的逻辑就介绍到这里,如果没有选择到合适的 Node,则针对该流控规则,默认放行。

2.4.1.2 TrafficShapingController canPass

经过上一个步骤获取到对应的实时统计数据,接下来就是根据数据与流控规则,是否匹配。Sentinel 中用于实现流控规则的匹配其类体系如图所示:

由于篇幅的关系,本节只会以 DefaultController 来介绍其实现原理,对应【流控模式:快速失败】,由于篇幅的关系,其他两种流控模式将在下文详细探讨。
DefaultController#canPasspublic boolean canPass(Node node, int acquireCount, boolean prioritized) {
    int curCount = avgUsedTokens(node);     // @1
    if (curCount + acquireCount > count) {   // @2
        if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {   // @3
            long currentTime;
            long waitInMs;
            currentTime = TimeUtil.currentTimeMillis();
            waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);   // @4
            if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {             // @5
                node.addWaitingRequest(currentTime + waitInMs, acquireCount);
                node.addOccupiedPass(acquireCount);
                sleep(waitInMs);                                                                                  // @6
                // PriorityWaitException indicates that the request will pass after waiting for {@link @waitInMs}.
                throw new PriorityWaitException(waitInMs);   // @7
            }
        }
        return false;     // @8
    }
    return true;       // @9
}

代码@1:首先先解释一下两个局部变量的含义:

  • int curCount
    当前已消耗的令牌数量,即当前时间窗口内已创建的线程数量(FLOW_GRADE_THREAD) 或已通过的请求个数(FLOW_GRADE_QPS)。

  • double count
    流控规则中配置的阔值(即一个时间窗口中总的令牌个数)

代码@2:如果当前请求的令牌数加上已消耗的令牌数之和小于总令牌数,则直接返回true,表示通过,见代码@9;如果当前时间窗口剩余令牌数小于需要申请的令牌数,则需要根据是否有优先级进行不同的处理。

  • 如果该请求存在优先级,即 prioritized 设置为 true,并且流控类型为基于QPS进行限流,则进入相关的处理逻辑,见代码@3~@8。

  • 否则直接返回 false,最终会直接抛出 FlowException,即快速失败,应用方可以捕捉该异常,对其业务进行容错处理。

代码@4:尝试抢占下一个滑动窗口的令牌,并返回该时间窗口所剩余的时间,如果获取失败,则返回 OccupyTimeoutProperty.getOccupyTimeout() 值,该返回值的作用就是当前申请资源的线程将 sleep(阻塞)的时间。

代码@5:如果 waitInMs 小于抢占的最大超时时间,则在下一个时间窗口中增加对应令牌数,并且线程将sleep,见代码@6。

代码@7:这里不是很明白为什么等待 waitMs 之后,还需要抛出 PriorityWaitException,那这个prioritized 机制、可抢占下一个时间窗口的令牌有什么意义呢?应该是一个BUG吧。

3、总结


整个 FlowSlot 限流规则就介绍到这里了,为了更加直观的认识其限流的流程,下面给出一张流程图来对上面的源码分析进行一个总结。

该篇注重理论与实践相结合,在进行源码解读之前先从流控规则配置界面入手,代入感比较强,文章再提供一张流程图。

整个限流部分目前还有所欠缺的两个部分:
1、流程规则的存储与加载。
2、其他几种流控后行为(预热、匀速排队等实现原理)

该部分内容将在后续文章中详细介绍,本文疑似发现一个BUG,也请大家一起交流、探讨。在分析 DefaultController canPass 方法时,prioritized 为 true 时,执行 sleep 方法唤醒后不管三七二十一,直接抛出 PriorityWaitException 这是要起到一个什么作用呢?

点【在看】与转发是一种美德,期待您的认可与鼓励,越努力越幸运。


欢迎加入我的知识星球,一起交流源码,探讨架构,打造高质量的技术交流圈,长按如下二维码


中间件兴趣圈 知识星球 正在对如下话题展开如火如荼的讨论:


1、【让天下没有难学的Netty-网络通道篇】

1、Netty4 Channel概述(已发表
2、Netty4 ChannelHandler概述(已发表
3、Netty4事件处理传播机制(已发表
4、Netty4服务端启动流程(已发表
5、Netty4 NIO 客户端启动流程
6、Netty4 NIO线程模型分析
7、Netty4编码器、解码器实现原理
8、Netty4 读事件处理流程
9、Netty4 写事件处理流程
10、Netty4 NIO Channel其他方法详解

2、Java 并发框架(JUC) 探讨【面试神器
3、源码分析Alibaba Sentienl 专栏背后的写作与学习技巧

您可能也对以下帖子感兴趣

文章有问题?点此查看未经处理的缓存