当前位置：首页 > news >正文

sentinel学习笔记6-限流降级（上）

news 2025/7/4 4:05:32

本文属于sentinel学习笔记系列。网上看到吴就业老师的专栏，写的好值得推荐，我整理的有所删减，推荐看原文。

https://blog.csdn.net/baidu_28523317/category_10400605.html

sentinel 实现限流降级、熔断降级、黑白名单限流降级、系统自适应限流降级以及热点参数限流降级都是由 ProcessorSlot、Checker、Rule、RuleManager 组合完成。ProcessorSlot 作为调用链路的切入点，负责调用 Checker 检查当前请求是否可以放行；Checker 则根据资源名称从 RuleManager 中拿到为该资源配置的 Rule（规则），取 ClusterNode 统计的实时指标数据与规则对比，如果达到规则的阈值则抛出 Block 异常，抛出 Block 异常意味着请求被拒绝，也就实现了限流或熔断。

可以总结为以下三个步骤：

在 ProcessorSlot#entry 方法中调用 Checker#check 方法，并将 DefaultNode 传递给 Checker。
Checker 从 DefaultNode 拿到 ClusterNode，并根据资源名称从 RuleManager 获取为该资源配置的规则。
Checker 从 ClusterNode 中获取当前时间窗口的某项指标数据（QPS、avgRt 等）与规则的阈值对比，如果达到规则的阈值则抛出 Block 异常（也有可能将 check 交给 Rule 去实现）。

限流规则与规则配置加载器

rule

规则是围绕资源配置的，接口Rule 只定义获取资源。

public interface Rule {/*** Get target resource of this rule.** @return target resource of this rule*/String getResource();}

public abstract class AbstractRule implements Rule {/*** rule id. */private Long id;/*** Resource name. 资源名*/private String resource;/***  流控对应的调用来源*/private String limitApp;

Rule、AbstractRule 与其它实现类的关系如下图所示：

FlowRule 是限流规则配置类，FlowRule 继承 AbstractRule 并实现 Rule 接口。FlowRule 源码如下

public class FlowRule extends AbstractRule {public FlowRule() {super();setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);}public FlowRule(String resourceName) {super();setResource(resourceName);setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);}/*** The threshold type of flow control (0: thread count, 1: QPS).* 限流阈值类型*/private int grade = RuleConstant.FLOW_GRADE_QPS;/*** Flow control threshold count. 限流阈值 配置的是qps类型则代表qps的值；配置的是线程数类型则代表线程数*/private double count;/*** Flow control strategy based on invocation chain.* 流控限流策略* {@link RuleConstant#STRATEGY_DIRECT} for direct flow control (by origin);* {@link RuleConstant#STRATEGY_RELATE} for relevant flow control (with relevant resource);* {@link RuleConstant#STRATEGY_CHAIN} for chain flow control (by entrance resource).*/private int strategy = RuleConstant.STRATEGY_DIRECT;/*** Reference resource in flow control with relevant resource or context.* 关联流控的资源*/private String refResource;/*** Rate limiter control behavior.  流控效果控制* 0. default(reject directly), 1. warm up, 2. rate limiter, 3. warm up + rate limiter*/private int controlBehavior = RuleConstant.CONTROL_BEHAVIOR_DEFAULT;//对应流控效果为Warm Up情况下，冷启动时长（预热时长），单位秒private int warmUpPeriodSec = 10;/*** Max queueing time in rate limiter behavior.* 对应流控效果为排队等待情况下，出现的超时时间*/private int maxQueueingTimeMs = 500;// 对应新增流控规则页面的是否集群private boolean clusterMode;/*** Flow rule config for cluster mode.集群流控的相关配置*/private ClusterFlowConfig clusterConfig;/*** The traffic shaping (throttling) controller.  流量整形的实现，不同流控效果有不同算法*/private TrafficShapingController controller;

字段属性有些多，可以对比sentinel 限流保护-笔记-CSDN博客跟官网文档来理解

RuleManager

Sentinel 中用来管理规则配置的类都以规则类的名称+Manger 命名，用来加载限流规则配置以及缓存限流规则配置的类为 FlowRuleManager，其部分源码如下：

public class FlowRuleManager {// 缓存限流规则private static volatile Map<String, List<FlowRule>> flowRules = new HashMap<>();// PropertyListener 监听器private static final FlowPropertyListener LISTENER = new FlowPropertyListener();//SentinelProperty ,默认的 DynamicSentinelPropertyprivate static SentinelProperty<List<FlowRule>> currentProperty = new DynamicSentinelProperty<List<FlowRule>>();/** the corePool size of SCHEDULER must be set at 1, so the two task ({@link #startMetricTimerListener()} can run orderly by the SCHEDULER **/@SuppressWarnings("PMD.ThreadPoolCreationRule")private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(1,new NamedThreadFactory("sentinel-metrics-record-task", true));static {//给默认的 SentinelProperty 注册监听器（FlowPropertyListener）currentProperty.addListener(LISTENER);startMetricTimerListener();}。。。public static List<FlowRule> getRules() {List<FlowRule> rules = new ArrayList<FlowRule>();for (Map.Entry<String, List<FlowRule>> entry : flowRules.entrySet()) {rules.addAll(entry.getValue());}return rules;}//更新规则public static void loadRules(List<FlowRule> rules) {currentProperty.updateValue(rules);}

我们之前demo对用FlowRuleManager.loadRules()来更新规则生效，注意1.8.6版本这里面

SentinelProperty ，SentinelProperty 是 Sentinel 提供的一个接口，可注册到 Sentinel 提供的各种规则的 Manager，例如 FlowRuleManager，并且可以给 SentinelProperty 添加监听器，在配置改变时，你可以调用 SentinelProperty#updateValue 方法，由它负责调用监听器去更新规则，而不需要调用 FlowRuleManager#loadRules 方法。这块先先不展开梳理，后面结合nacos再看。

限流处理器插槽：FlowSlot

FlowSlot 是实现限流功能的切入点，它作为 ProcessorSlot 插入到 ProcessorSlotChain 链表中，在 entry 方法中调用 Checker 去判断是否需要拒绝当前请求，如果需要拒绝请求则抛出 Block 异常。FlowSlot 的源码如下：

@Spi(order = Constants.ORDER_FLOW_SLOT)
public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {private final FlowRuleChecker checker;public FlowSlot() {this(new FlowRuleChecker());}/*** Package-private for test.** @param checker flow rule checker* @since 1.6.1*/FlowSlot(FlowRuleChecker checker) {AssertUtil.notNull(checker, "flow checker should not be null");this.checker = checker;}@Overridepublic 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);}@Overridepublic void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {fireExit(context, resourceWrapper, count, args);}// 规则生产者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);}};
}

限流规则检查器：FlowRuleChecker

FlowRuleChecker 负责判断是否需要拒绝当前请求，方法很多，先看看调用checkFlow

public class FlowRuleChecker {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);}}}}

注意：遍历限流规则，只要有一个限流规则达到限流阈值即可抛出 FlowException，使用 FlowException 目的是标志当前请求因为达到限流阈值被拒绝，FlowException 是 BlockException 的子类；

canPassCheck 方法返回 true 说明允许请求通过，反之则不允许通过。canPassCheck 方法源码如下：

    public boolean canPassCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node, int acquireCount,boolean prioritized) {String limitApp = rule.getLimitApp();//当前限流规则只对哪个调用来源生效,一般不为null,默认为“default”（不限定调用来源）if (limitApp == null) {return true;}// 集群模式下的规则检测if (rule.isClusterMode()) {return passClusterCheck(rule, context, node, acquireCount, prioritized);}//单机模式下规则检测return passLocalCheck(rule, context, node, acquireCount, prioritized);}

先不讨论集群限流的情况，看看单机的passLocalCheck

    private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,boolean prioritized) {//根据调用来源和“调用关系限流策略”选择 DefaultNode；Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);if (selectedNode == null) {return true;}return rule.getRater().canPass(selectedNode, acquireCount, prioritized);}

selectNodeByRequesterAndStrategy返回ClusterBuilderSlot阶段生成的ClusterNode
getRater 返回TrafficShapingController，在默认模式下返回流控效果策略DefaultController。DefaultController#canPass 完成canPassCheck。

下面分别看看这两个方法。

流控节点选择

selectNodeByRequesterAndStrategy 方法的实现有多种情况。原码如下：

public class FlowRuleChecker {static Node selectNodeByRequesterAndStrategy(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node) {// The limit app should not be empty.限流规则针对哪个来源生效String limitApp = rule.getLimitApp();// 基于调用关系的限流策略int strategy = rule.getStrategy();// 远程来源String origin = context.getOrigin();if (limitApp.equals(origin) && filterOrigin(origin)) {if (strategy == RuleConstant.STRATEGY_DIRECT) {// 1 Matches limit origin, return origin statistic node.return context.getOriginNode();}//2return selectReferenceNode(rule, context, node);} else if (RuleConstant.LIMIT_APP_DEFAULT.equals(limitApp)) {if (strategy == RuleConstant.STRATEGY_DIRECT) {//3 Return the cluster node.return node.getClusterNode();}//4return selectReferenceNode(rule, context, node);} else if (RuleConstant.LIMIT_APP_OTHER.equals(limitApp)&& FlowRuleManager.isOtherOrigin(origin, rule.getResource())) {if (strategy == RuleConstant.STRATEGY_DIRECT) {// 5return context.getOriginNode();}//6return 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;}if (strategy == RuleConstant.STRATEGY_RELATE) {return ClusterBuilderSlot.getClusterNode(refResource);}if (strategy == RuleConstant.STRATEGY_CHAIN) {if (!refResource.equals(context.getName())) {return null;}return node;}// No node.return null;}

如果当前限流规则的 limitApp 为 default，则说明该限流规则对任何调用来源都生效，针对所有调用来源限流，否则只针对指定调用来源限流。

1 如果调用来源与当前限流规则的 limitApp 相等，且 strategy 为 STRATEGY_DIRECT，则使用调用来源的 StatisticNode，实现针对调用来源限流。

2 前置条件与（1）相同，依然是针对来源限流。selectReferenceNode

strategy 为 STRATEGY_RELATE：使用引用资源的 ClusterNode；
strategy 为 STRATEGY_CHAIN：使用当前资源的 DefauleNode。

3当 limitApp 为 default 时，针对所有来源限流。如果 strategy 为 STRATEGY_DIRECT，则使用当前资源的 ClusterNode。

4 前置条件与（3）相同，依然是针对所有来源限流。selectReferenceNode

5 如果 limitApp 为 other，且该资源的所有限流规则都没有针对当前的调用来源限流。如果 strategy 为 STRATEGY_DIRECT，则使用 origin 的 StatisticNode。

6 前置条件与（5）一样。selectReferenceNode

从 selectNodeByRequesterAndStrategy 方法可以看出，Sentinel 之所以针对每个资源统计访问来源的指标数据，也是为了实现对丰富的限流策略的支持.比如针对调用来源限流可限制并发量较高的来源服务的请求，而对并发量低的来源服务的请求可不限流，或者是对一些并没有那么重要的来源服务限流。

TrafficShapingController

Sentinel 支持对超出限流阈值的流量采取效果控制器控制这些流量，流量效果控制支持：直接拒绝、Warm Up（冷启动）、匀速排队。对应 FlowRule 中的 controlBehavior 字段。在调用 FlowRuleManager#loadRules 方法时，FlowRuleManager 会将限流规则配置的 controlBehavior 转为对应的 TrafficShapingController。

public interface TrafficShapingController {/*** Check whether given resource entry can pass with provided count.* 判断当前请求是否能通过* @param node resource node* @param acquireCount count to acquire* @param prioritized whether the request is prioritized* @return true if the resource entry can pass; false if it should be blocked*/boolean canPass(Node node, int acquireCount, boolean prioritized);/*** Check whether given resource entry can pass with provided count.* 判断当前请求是否能通过* @param node resource node* @param acquireCount count to acquire* @return true if the resource entry can pass; false if it should be blocked*/boolean canPass(Node node, int acquireCount);
}

DefaultController

DefaultController 是默认使用的流量效果控制器，直接拒绝超出阈值的请求。当 QPS 超过限流规则配置的阈值，新的请求就会被立即拒绝，抛出 FlowException。

  @Overridepublic boolean canPass(Node node, int acquireCount, boolean prioritized) {// 获取当前已使用的token：qps 算每秒被放行的请求数，threads 统计的当前并行占用的线程数int curCount = avgUsedTokens(node);//当前已使用token + 获取的token 大于token数量的场景if (curCount + acquireCount > count) {//qps 且prioritized 参数的值为 true(有优先级的请求可以占用未来时间窗口的统计指标)if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {long currentTime;long waitInMs;currentTime = TimeUtil.currentTimeMillis();//当前请求需要等待的时间，单位毫秒waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {//将休眠之后对应的时间窗口的 pass(通过)这项指标数据的值加上 acquireCountnode.addWaitingRequest(currentTime + waitInMs, acquireCount);// 添加占用未来的 pass 指标的数量node.addOccupiedPass(acquireCount);// 休眠等待，当前线程阻塞sleep(waitInMs);// PriorityWaitException indicates that the request will pass after waiting for {@link @waitInMs}.//休眠结束后抛出 PriorityWait 异常，表示当前请求是等待了 waitInMs 之后通过的throw new PriorityWaitException(waitInMs);}}return false;}return true;}private int avgUsedTokens(Node node) {if (node == null) {return DEFAULT_AVG_USED_TOKENS;}return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());}

一般情况下，prioritized 参数的值为 false，所以这个 canPass 方法实现的流量效果就是直接拒绝。

限于篇幅，其他限流RateLimiterController、WarmUpController、WarmUpRateLimiterController 待整理。

查看全文

http://www.lryc.cn/news/510516.html