Sentinel源码分析----流控规则与FlowSlot

2019-03-03 本文已影响0人 _六道木

FlowSlot主要是用来进行流控规则的处理，直接看下代码

    @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 {
        // 获取流控规则
        Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();
       //通过资源名称来获取规则列表
        List<FlowRule> rules = flowRules.get(resource.getName());
        if (rules != null) {
            for (FlowRule rule : rules) {
                // 遍历规则进行处理
                if (!canPassCheck(rule, context, node, count, prioritized)) {
                    // 如果规则校验不通过，那么抛出FlowException异常
                    throw new FlowException(rule.getLimitApp(), rule);
                }
            }
        }
    }

    boolean canPassCheck(FlowRule rule, Context context, DefaultNode node, int count, boolean prioritized) {
        // 交由FlowRuleChecker进行逻辑处理
        return FlowRuleChecker.passCheck(rule, context, node, count, prioritized);
    }

这里的flowRules是一个全量的规则列表，例如我在控制台配置了如下的规则：

image.png

那么flowRules中就有两个元素，key分别是test和hello，对应的值是一个集合，集合里只有一个元素，就是实际的规则实体FlowRule，具体值与我们配置相关，看下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);
    }
    private int grade = RuleConstant.FLOW_GRADE_QPS;
    private double count;
    private int strategy = RuleConstant.STRATEGY_DIRECT;
    private String refResource;
    private int controlBehavior = RuleConstant.CONTROL_BEHAVIOR_DEFAULT;

    private int warmUpPeriodSec = 10;

    private int maxQueueingTimeMs = 500;

    private boolean clusterMode;

    private ClusterFlowConfig clusterConfig;

    private TrafficShapingController controller;
}

limitApp：对应新增流控规则页面的来源应用
resource：对应新增流控规则页面的资源名
grade：对应新增流控规则页面的阈值类型
count：如果页面配置的是qps类型，字段则代表qps的值；如果配置的是线程数类型，字段则代表线程数
strategy：对应新增流控规则页面的流控模式
refResource：对应流控策略为关联情况下，出现的关联资源 或对应流控策略为链路情况下，出现的入口资源
controlBehavior：对应新增流控规则页面的流控效果
warmUpPeriodSec：对应流控效果为Warm Up情况下，出现的预热时长
maxQueueingTimeMs：对应流控效果为排队等待情况下，出现的超时时间
clusterMode：对应新增流控规则页面的是否集群
ClusterFlowConfig：集群流控的相关配置
TrafficShapingController：流量整形的实现，不同流控效果有不同算法

FlowRule和页面配置的规则一一对应，通过控制台配置后可以将这些值推送到机器上生成对应的FlowRule

接下来看下FlowRuleChecker.passCheck对具体规则的处理

    static boolean passCheck(/*@NonNull*/ 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);//3
    }

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

        return rule.getRater().canPass(selectedNode, acquireCount);//5
    }

标记1：limitApp是页面上的来源应用，默认是default，表示代表所有的应用，这里如果为空则默认通过，因为代码中约定了default是代表所有应用，所以空值为非法值，这里想了一下为什么不把default或者空值当做代表所有应用的限流，可能是因为空值还包括规则字段丢失的情况，应该算作异常情况
标记2：集群模式特殊处理，这里暂不考虑，后续分析
标记3：本地限流逻辑实现
标记4：根据不同情况选择不同Node(这里会涉及上篇文章的知识点)
标记5：根据不同情况调用不同TrafficShapingController实现进行判断

节点选择

上篇文章中分析了Sentinel的各种Node的含义，为什么要设计那么多种类型呢？下面就会看到，对于不同的流控规则而言，需要去拿不同的Node来获取统计的数据，具体看代码（对于各种Node的知识点这里不再详细分析，具体看下上篇文章）

    static Node selectNodeByRequesterAndStrategy(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)) {//1
            if (strategy == RuleConstant.STRATEGY_DIRECT) {
                return context.getOriginNode();// 2
            }

            return selectReferenceNode(rule, context, node);//3
        } else if (RuleConstant.LIMIT_APP_DEFAULT.equals(limitApp)) {//4
            if (strategy == RuleConstant.STRATEGY_DIRECT) {//5
                return node.getClusterNode();
            }

            return selectReferenceNode(rule, context, node);//6
        } else if (RuleConstant.LIMIT_APP_OTHER.equals(limitApp)
            && FlowRuleManager.isOtherOrigin(origin, rule.getResource())) {//7
            if (strategy == RuleConstant.STRATEGY_DIRECT) {
                return context.getOriginNode();//8
            }

            return selectReferenceNode(rule, context, node);//9
        }

        return null;
    }

标记1：如果流控规则配置了来源应用且不是"default"或者"other"这种特殊值，那么这种时候该规则就只对配置的来源应用生效，例如：配置了A应用对test资源qps为10，那么就要先取到当前A应用的qps看看是否超过10
标记2：如果是直接限流类型，那么也就是上面举的栗子，获取A应用的统计数据，即A应用对应的OriginNode进行判断
标记:3/6/9：selectReferenceNode方法是对流控模式为关联或者链路的处理
标记4：这种情况limitApp是"default"，代表针对所有应用
标记5：如果是直接限流类型，因为不是针对某个应用进行限流，所以就需要取当前资源的ClusterNode节点，因为ClusterNode表示所有应用对该资源的所有请求情况
标记7：这个是"other"值的处理，假设当前请求来源不在当前规则的limitApp中，则进行下面的处理
标记8：如果是直接限流类型，则返回OriginNode

关于7的应用，具体栗子，假设一个资源有如下规则，属性如下

ruleName	limitApp
rule1	A
rule2	default
rule3	C
rule4	other

那么rule4只会处理来源应用非A、C、default的应用，例如D，E等统一使用rule4这个规则，这种情况实际应用场景是：假设有非常多的来源应用，但是又不能统一使用某个规则，因为可能某个来源应用的请求量很大，统一使用某个规则会导致请求量小的应用被影响；又不能每个来源应用配置一个规则，那这样会配到手抖，那么可以为ABC分别配置一个规则(假设ABC是请求量非常大的，和其他的差别很大)，然后再配置一个other，这样其他请求量小的就可以使用这个规则了

流控模式

关联与链路这两种模式在wiki的介绍中，统一被称为基于调用关系的流量控制

流控模式：关联

当两个资源之间具有资源争抢或者依赖关系的时候，这两个资源便具有了关联。比如对数据库同一个字段的读操作和写操作存在争抢，读的速度过高会影响写得速度，写的速度过高会影响读的速度。如果放任读写操作争抢资源，则争抢本身带来的开销会降低整体的吞吐量。可使用关联限流来避免具有关联关系的资源之间过度的争抢，举例来说，read_db 和 write_db 这两个资源分别代表数据库读写，我们可以给 read_db 设置限流规则来达到写优先的目的：设置 FlowRule.strategy 为 RuleConstant.RELATE 同时设置 FlowRule.ref_identity 为 write_db。这样当写库操作过于频繁时，读数据的请求会被限流。

gayhub的wiki上描述如上，也就是read_db的请求量会被write_db影响，假设read_db配置的规则如下：

image.png

这种情况下，假设write_db没有被执行，那么read_db最大能到多少的qps？

看下代码

    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) {// 1
            return ClusterBuilderSlot.getClusterNode(refResource);
        }

        //....链路模式的处理
        return null;
    }

看到标记1的地方，关联流控模式是使用关联资源即refResource去获取资源的ClusterNode，以write_db和read_db为例，当read_db请求的时候，是把write_db的ClusterNode与规则进行比较，那么上面的问题就会有答案了，假设write_db一直没有请求，那么read_db就没有限制，因为write_db的ClusterNode数据为空

流控模式：链路

                  machine-root
                    /       \
                   /         \
             Entrance1     Entrance2
                /             \
               /               \
      DefaultNode(nodeA)   DefaultNode(nodeA)

如上所示的Node分布情况，资源nodeA分别在两个上下文Entrance1和Entrance2下进行调用，假设在上下文Entrance1的调用量很大，而上下文Entrance2的调用量很小，我们想针对Entrance1上下文的nodeA调用进行限流，那么可以使用链路限流模式，配置如下：

image.png
那么在上下文Entrance2下对nodeA的调用就没有影响，看下代码

    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_CHAIN) {//2
            if (!refResource.equals(context.getName())) {
                return null;
            }
            return node;
        }
        return null;
    }

发现当前上下文(context.getName())如果和配置(refResource)的不一样，则返回null，外部如果返回的Node为null，则直接返回true了，那么Entrance2在这种情况下就直接通过了

流控效果

当节点选择完毕后，调用rule.getRater().canPass(selectedNode, acquireCount)开始执行判断，getRater()返回的是TrafficShapingController的实现类，根据不同流控效果有不同的实现

    //com.alibaba.csp.sentinel.slots.block.flow.FlowRuleUtil#generateRater
    private static TrafficShapingController generateRater(/*@Valid*/ FlowRule rule) {
        if (rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {
            switch (rule.getControlBehavior()) {
                case RuleConstant.CONTROL_BEHAVIOR_WARM_UP:
                    return new WarmUpController(rule.getCount(), rule.getWarmUpPeriodSec(),
                        ColdFactorProperty.coldFactor);
                case RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER:
                    return new RateLimiterController(rule.getMaxQueueingTimeMs(), rule.getCount());
                case RuleConstant.CONTROL_BEHAVIOR_WARM_UP_RATE_LIMITER:
                    return new WarmUpRateLimiterController(rule.getCount(), rule.getWarmUpPeriodSec(),
                        rule.getMaxQueueingTimeMs(), ColdFactorProperty.coldFactor);
                case RuleConstant.CONTROL_BEHAVIOR_DEFAULT:
                default:
                    // Default mode or unknown mode: default traffic shaping controller (fast-reject).
            }
        }
        return new DefaultController(rule.getCount(), rule.getGrade());
    }

快速失败

快速失败这种情况，使用的是DefaultController，也是最简单的一个

    public boolean canPass(Node node, int acquireCount, boolean prioritized) {
        int curCount = avgUsedTokens(node);
        if (curCount + acquireCount > count) {
            return false;
        }
        return true;
    }
    private int avgUsedTokens(Node node) {
        if (node == null) {
            return -1;
        }
        return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)node.passQps();
    }

获取当前token数，和当前请求的数量相加，看看是否大于规则配置的值

排队等待

当页面流控效果选择排队等待的时候，会出现超时时间的选项，该效果是让请求匀速的通过，可用于消息队列在消费的时候对流量的控制，对应的是漏桶算法，算法实现的代码是RateLimiterController

    public boolean canPass(Node node, int acquireCount, boolean prioritized) {
        if (acquireCount <= 0) {
            return true;
        }
        if (count <= 0) {
            return false;
        }

        long currentTime = TimeUtil.currentTimeMillis();

        long costTime = Math.round(1.0 * (acquireCount) / count * 1000);//1

        long expectedTime = costTime + latestPassedTime.get();//2

        if (expectedTime <= currentTime) {//3
            latestPassedTime.set(currentTime);//4
            return true;
        } else {//5
            long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();//6
            if (waitTime > maxQueueingTimeMs) {//7
                return false;
            } else {//8
                long oldTime = latestPassedTime.addAndGet(costTime);//9
                try {
                    waitTime = oldTime - TimeUtil.currentTimeMillis();//10
                    if (waitTime > maxQueueingTimeMs) {//11
                        latestPassedTime.addAndGet(-costTime);//12
                        return false;
                    }
                    if (waitTime > 0) {//13
                        Thread.sleep(waitTime);
                    }
                    return true;
                } catch (InterruptedException e) {
                }
            }
        }
        return false;
    }

这里将上面几个时间用变量表示

下一次能获取到token通过的时间为t1
每次请求花费时间为t2
上一次获取到token通过的时间为t3

代码每个标记意义如下：

1：即当前请求token数(默认是1)/qps1000=1/qps1000=一个请求需要花费多少时间，假设设置的qps为10，即一秒允许有10个请求通过，那么每个请求的时间就是1/10*1000=100毫秒
2：t3+t2=t1
3：如果当前时间已经在t1后面了，那么请求可以被通过
4：通过的时候需要重设一下t3
5：代表当前请求到来的时候，还没到达能够t1
6：t2+t3-当前时间=代表离t1还差多少时间
7：如果请求很多，每个都需要进行排队，那么会导致越后面的请求等待的时候会更久，那么当时间超过设置的最大间隔，则返回false直接失败
8：这里表示未到达最大的间隔
9：更新t3并返回最新值(这个的变量命名感觉有点问题，addAndGet返回的是最新值)
10~11：在7已经判断过一次了，这里又判断一次的原因是因为可能有多个线程并发执行的时候，在7的时候还未超过最大的间隔时间，而经过latestPassedTime.addAndGet的处理之后，可能有某些线程已经超过了这个时间，所以这里又判断了一遍
12：和7不一样，7只是进行运算，这里是先更新了t3了，所以需要减回去
13：得到等待时间后，使用Sleep是线程睡眠一定时间

Warm Up

还有一种流控效果是Warm Up，该算法类似于令牌桶算法，其代码与Guava的RateLimiter原理类似，限于个人能力，没能看懂其中原理。。。。