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

标签：node 降级 return rule 限流 private context sentinel

本文属于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 ,默认的 DynamicSentinelProperty
    private 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;
    }

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

    @Override
    public 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();
            }
                 //2
            return 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();
            }
            //4
            return selectReferenceNode(rule, context, node);
        } else if (RuleConstant.LIMIT_APP_OTHER.equals(limitApp)
            && FlowRuleManager.isOtherOrigin(origin, rule.getResource())) {
            if (strategy == RuleConstant.STRATEGY_DIRECT) {
                // 5
                return context.getOriginNode();
            }
              //6
            return 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。

  @Override
    public 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(通过)这项指标数据的值加上 acquireCount
                    node.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 待整理。

标签：node,降级,return,rule,限流,private,context,sentinel
From： https://blog.csdn.net/bohu83/article/details/144658597