前言

周末花了 2 天时间学习了额 RabbitMQ，总结了最核心的知识点，带大家快速掌握 RabbitMQ，整理不易希望帮忙点赞，转发，分享下，谢谢

代码 @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#canPass

public 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 这是要起到一个什么作用呢？

点赞是一种美德，麻烦帮忙点个赞，谢谢

欢迎加笔者微信号(dingwpmz)，加群探讨，笔者优质专栏目录：

1、[源码分析 RocketMQ 专栏(40 篇+)](()

2、[源码分析 Sentinel 专栏(12 篇+)](()

3、[源码分析 Dubbo 专栏(28 篇+)](()

4、[源码分析 Mybatis 专栏](()

5、[源码分析 Netty 专栏(18 篇+)](()

《一线大厂 Java 面试真题解析+Java 核心总结学习笔记+最新全套讲解视频+实战项目源码》开源

Java 优秀开源项目：

• ali1024.coding.net/public/P7/Java/git

最后

由于篇幅原因，就不多做展示了