Ali-Sentinel-链路控制

归档

GitHub: Ali-Sentinel-链路控制

链结构

参考：入口控制-处理链

具体实现

NodeSelectorSlot

给上下文设置统计节点
com.alibaba.csp.sentinel.slots.nodeselector.NodeSelectorSlot

@Spi(isSingleton = false, order = Constants.ORDER_NODE_SELECTOR_SLOT)
public class NodeSelectorSlot extends AbstractLinkedProcessorSlot<Object> {

    private volatile Map<String, DefaultNode> map = new HashMap<String, DefaultNode>(10);

    @Override
    public void entry(
        Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args
    ) throws Throwable {
        DefaultNode node = map.get(context.getName());
        if (node == null) {
            synchronized (this) {   // DCL
                node = map.get(context.getName());
                if (node == null) {
                    node = new DefaultNode(resourceWrapper, null);              // 创建默认节点
                    HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());
                    cacheMap.putAll(map);
                    cacheMap.put(context.getName(), node);
                    map = cacheMap;                                             // COW
                    ((DefaultNode) context.getLastNode()).addChild(node);       // 构建调用树  ref: sign_m_010 | sign_m_020
                }
            }
        }

        context.setCurNode(node);                                               // 保存到上下文  ref: sign_m_011
        fireEntry(context, resourceWrapper, node, count, prioritized, args);    // (将统计节点) 流转给下游
    }
}

com.alibaba.csp.sentinel.context.Context
- entranceNode 在上下文-创建 sign_m_020 中赋值

    // sign_m_010 获取尾节点
    public Node getLastNode() {
        if (curEntry != null && curEntry.getLastNode() != null) {
            return curEntry.getLastNode();  // sign_m_040
        } else {
            return entranceNode;    // 一般返回此
        }
    }

    // sign_m_011
    public Context setCurNode(Node node) {
        this.curEntry.setCurNode(node); //  sign_m_030
        return this;
    }

com.alibaba.csp.sentinel.node.DefaultNode

    // sign_m_020 添加子节点
    public void addChild(Node node) {
        ... // 省略 node 空判断
        if (!childList.contains(node)) {
            synchronized (this) {       // DCL
                if (!childList.contains(node)) {
                    Set<Node> newSet = new HashSet<>(childList.size() + 1);
                    newSet.addAll(childList);
                    newSet.add(node);
                    childList = newSet; // COW
                }
            }
        }
    }

com.alibaba.csp.sentinel.Entry


    public abstract Node getLastNode();

    // sign_m_030
    public void setCurNode(Node node) {
        this.curNode = node;
    }

    // sign_m_031
    public Node getCurNode() {
        return curNode;
    }

com.alibaba.csp.sentinel.CtEntry

    // sign_m_040
    @Override
    public Node getLastNode() {
        return parent == null ? null : parent.getCurNode(); // sign_m_031
    }

ClusterBuilderSlot

给统计节点设置集群(统计)节点
- 只是用于统计
com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot

@Spi(isSingleton = false, order = Constants.ORDER_CLUSTER_BUILDER_SLOT)
public class ClusterBuilderSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    private static volatile Map<ResourceWrapper, ClusterNode> clusterNodeMap = new HashMap<>(); // 记录所有的节点
    private static final Object lock = new Object();
    private volatile ClusterNode clusterNode = null;

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable 
    {
        if (clusterNode == null) {
            synchronized (lock) {   // DCL
                if (clusterNode == null) {
                    clusterNode = new ClusterNode(resourceWrapper.getName(), resourceWrapper.getResourceType());
                    HashMap<ResourceWrapper, ClusterNode> newMap = new HashMap<>(Math.max(clusterNodeMap.size(), 16));
                    newMap.putAll(clusterNodeMap);
                    newMap.put(node.getId(), clusterNode);
                    clusterNodeMap = newMap;    // COW
                }
            }
        }
        node.setClusterNode(clusterNode);       // 设置集群节点

        ... // 省略设置源节点处理

        fireEntry(context, resourceWrapper, node, count, prioritized, args);    // 传递给下游节点
    }

}

LogSlot

日志异常记录
com.alibaba.csp.sentinel.slots.logger.LogSlot

@Spi(order = Constants.ORDER_LOG_SLOT)
public class LogSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode obj, int count, boolean prioritized, Object... args)
        throws Throwable 
    {
        try {
            fireEntry(context, resourceWrapper, obj, count, prioritized, args); // 先传给下游
        } catch (BlockException e) {
            EagleEyeLogUtil.log(resourceWrapper.getName(), e.getClass().getSimpleName(), e.getRuleLimitApp(),
                context.getOrigin(), e.getRule().getId(), count);
            throw e; // 继续往上抛
        } catch (Throwable e) {
            // 下游处理出错，则记录异常日志
            RecordLog.warn("Unexpected entry exception", e);
        }
    }
}

StatisticSlot

统计各种数据
com.alibaba.csp.sentinel.slots.statistic.StatisticSlot

@Spi(order = Constants.ORDER_STATISTIC_SLOT)
public class StatisticSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    @Override // sign_m_400 记录通过数
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable 
    {
        try {
            /**
             * 先传给下游处理，让下游先处理，
             * 后面自己才做统计，
             * 这样，下游要处理 QPS 等，要等下次才有数据
             */
            fireEntry(context, resourceWrapper, node, count, prioritized, args);

            // 添加线程计数和通过计数
            node.increaseThreadNum();
            node.addPassRequest(count);                     // ref: sign_m_401

            ... // 省略入口源节点添加计数

            // 全局入站节点添加计数 (其用于系统流控)
            if (resourceWrapper.getEntryType() == EntryType.IN) {
                Constants.ENTRY_NODE.increaseThreadNum();
                Constants.ENTRY_NODE.addPassRequest(count); // ref: sign_m_401
            }

            ... // sign_call_100 省略注册的回调器处理
        } catch (PriorityWaitException ex) {
            ... // 省略此异常处理 (相当于下游抛出此异常，上面的统计 (除 pass 外) 再走一次)
        } catch (BlockException e) {
            context.getCurEntry().setBlockError(e); // 记录异常
            node.increaseBlockQps(count);           // Add block count.
            
            ... // 省略入口源节点添加计数

            // 全局入站节点添加计数
            if (resourceWrapper.getEntryType() == EntryType.IN) {
                Constants.ENTRY_NODE.increaseBlockQps(count);
            }

            ... // 省略注册的回调器处理
            throw e;
        } catch (Throwable e) {
            context.getCurEntry().setError(e);      // 记录异常
            throw e;
        }
    }

    @Override // 记录完成数
    public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
        Node node = context.getCurNode();

        if (context.getCurEntry().getBlockError() == null) {
            // 计算响应时间 (当前时间 - 入口创建时间)
            long completeStatTime = TimeUtil.currentTimeMillis();
            context.getCurEntry().setCompleteTimestamp(completeStatTime);
            long rt = completeStatTime - context.getCurEntry().getCreateTimestamp();

            Throwable error = context.getCurEntry().getError();

            // 记录响应时间和成功次数
            recordCompleteFor(node, count, rt, error);
            recordCompleteFor(context.getCurEntry().getOriginNode(), count, rt, error);
            if (resourceWrapper.getEntryType() == EntryType.IN) {
                recordCompleteFor(Constants.ENTRY_NODE, count, rt, error);
            }
        }

        ... // sign_call_200 省略注册的回调器处理

        fireExit(context, resourceWrapper, count, args); // 传给下游
    }

    private void recordCompleteFor(Node node, int batchCount, long rt, Throwable error) {
        ... // 省略 node 为空返回处理
        node.addRtAndSuccess(rt, batchCount);       // 添加 RT 和完成数
        node.decreaseThreadNum();                   // 减线程数
        if (error != null && !(error instanceof BlockException)) {
            node.increaseExceptionQps(batchCount);  // 添加异常计数
        }
    }
}

com.alibaba.csp.sentinel.node.DefaultNode

    // sign_m_401
    @Override
    public void addPassRequest(int count) {
        super.addPassRequest(count);            // ref: sign_m_410
        this.clusterNode.addPassRequest(count); // ref: sign_m_410
    }

com.alibaba.csp.sentinel.node.StatisticNode
- Metric.addPass 方法参考：节点与度量-addpass

    // sign_m_410
    @Override
    public void addPassRequest(int count) {
        rollingCounterInSecond.addPass(count);  // 参考： 节点与度量-addpass
        rollingCounterInMinute.addPass(count);
    }

AuthoritySlot

黑、白名单权限校验
com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot

@Spi(order = Constants.ORDER_AUTHORITY_SLOT)
public class AuthoritySlot extends AbstractLinkedProcessorSlot<DefaultNode> {
    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)
        throws Throwable 
    {
        checkBlackWhiteAuthority(resourceWrapper, context);                     // 先校验  ref: sign_m_501
        fireEntry(context, resourceWrapper, node, count, prioritized, args);    // 再传给下游
    }

    // sign_m_501
    void checkBlackWhiteAuthority(ResourceWrapper resource, Context context) throws AuthorityException {
        Map<String, Set<AuthorityRule>> authorityRules = AuthorityRuleManager.getAuthorityRules();  // 获取总规则
        ... // 省略 authorityRules 为 null 返回

        Set<AuthorityRule> rules = authorityRules.get(resource.getName());  // 获取当前资源的规则
        ... // 省略 rules 为 null 返回

        // 一般一个资源只有一条规则  ref: AuthorityRuleManager.RulePropertyListener #loadAuthorityConf
        for (AuthorityRule rule : rules) {
            if (!AuthorityRuleChecker.passCheck(rule, context)) {           // 依次校验  ref: sign_m_510
                throw new AuthorityException(context.getOrigin(), rule);    // 校验不通过则抛异常
            }
        }
    }
}

com.alibaba.csp.sentinel.slots.block.authority.AuthorityRuleChecker

    // sign_m_510 校验权限规则
    static boolean passCheck(AuthorityRule rule, Context context) {
        String requester = context.getOrigin();

        ... // 省略 requester 和 rule.getLimitApp() 空判断

        int pos = rule.getLimitApp().indexOf(requester);
        boolean contain = pos > -1; // 包含

        // 加此判断可省略不包含时的多余处理
        // 下面的处理相当于：逗号分隔再依次精确匹配
        if (contain) {
            boolean exactlyMatch = false;
            String[] appArray = rule.getLimitApp().split(",");  // 英文逗号分隔
            for (String app : appArray) {
                if (requester.equals(app)) {    // 精确匹配
                    exactlyMatch = true;        // 匹配上，才算包含
                    break;
                }
            }
            contain = exactlyMatch;
        }

        int strategy = rule.getStrategy();
        if (strategy == RuleConstant.AUTHORITY_BLACK && contain) {  // 黑名单，包含：则不通过
            return false;
        }
        if (strategy == RuleConstant.AUTHORITY_WHITE && !contain) { // 白名单，不包含：则不通过
            return false;
        }
        return true;    // 通过
    }

SystemSlot

com.alibaba.csp.sentinel.slots.system.SystemSlot

@Spi(order = Constants.ORDER_SYSTEM_SLOT)
public class SystemSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable 
    {
        SystemRuleManager.checkSystem(resourceWrapper, count);                  // 先校验  ref: sign_m_610
        fireEntry(context, resourceWrapper, node, count, prioritized, args);    // 再传给下游
    }
}

com.alibaba.csp.sentinel.slots.system.SystemRuleManager

    // sign_m_610
    public static void checkSystem(ResourceWrapper resourceWrapper, int count) throws BlockException {
        ... // 省略 resourceWrapper 为 null 返回
        ... // 省略 checkSystemStatus 为 false 返回
        ... // 省略 资源类型 不为 入站 返回

        // total qps
        double currentQps = Constants.ENTRY_NODE.passQps(); // 在 StatisticSlot 里记录，ref: sign_m_400
        if (currentQps + count > qps) {
            throw new SystemBlockException(resourceWrapper.getName(), "qps");
        }

        ... // 省略 线程数 校验
        ... // 省略 平均RT 校验

        // load. BBR algorithm.
        if (highestSystemLoadIsSet && getCurrentSystemAvgLoad() > highestSystemLoad) {
            if (!checkBbr(currentThread)) { // ref: sign_m_611
                throw new SystemBlockException(resourceWrapper.getName(), "load");
            }
        }

        ... // 省略 CPU 校验 (CPU 数据每秒读取一次)
    }

    // sign_m_611
    private static boolean checkBbr(int currentThread) {
        if (currentThread > 1 &&
            currentThread > Constants.ENTRY_NODE.maxSuccessQps() * Constants.ENTRY_NODE.minRt() / 1000) {
            return false;
        }
        return true;
    }

FlowSlot

对当前资源进行流控
com.alibaba.csp.sentinel.slots.block.flow.FlowSlot
- 流控规则设置参考：入口控制-设置规则 sign_demo_020

@Spi(order = Constants.ORDER_FLOW_SLOT)
public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
    
    private final FlowRuleChecker 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);          // 先校验 sign_m_710
        fireEntry(context, resourceWrapper, node, count, prioritized, args);    // 再传给下游
    }

    // sign_m_710
    void checkFlow(ResourceWrapper resource, Context context, DefaultNode node, int count, boolean prioritized)
        throws BlockException 
    {
        checker.checkFlow(ruleProvider, resource, context, node, count, prioritized);   // sign_m_720
    }

    // 返回资源对应的流控规则集合
    private final Function<String, Collection<FlowRule>> ruleProvider = new Function<String, Collection<FlowRule>>() {
        @Override
        public Collection<FlowRule> apply(String resource) {
            Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();   // ref: sign_demo_020
            return flowRules.get(resource);
        }
    };
}

com.alibaba.csp.sentinel.slots.block.flow.FlowRuleChecker

    // sign_m_720
    public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,
                          Context context, DefaultNode node, int count, boolean prioritized) throws BlockException 
    {
        ... // 省略 ruleProvider 和 resource 为 null 返回
        Collection<FlowRule> rules = ruleProvider.apply(resource.getName());
        if (rules != null) {
            for (FlowRule rule : rules) {
                if (!canPassCheck(rule, context, node, count, prioritized)) {   // sign_m_721
                    throw new FlowException(rule.getLimitApp(), rule);
                }
            }
        }
    }

    // sign_m_721
    public boolean canPassCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                                    boolean prioritized) 
    {
        ... // 省略 rule.limitApp 为 null 返回 true
        if (rule.isClusterMode()) {
            // ref: 集群流控-流控原理-sign_m_411
            return passClusterCheck(rule, context, node, acquireCount, prioritized);
        }
        return passLocalCheck(rule, context, node, acquireCount, prioritized);  // sign_m_722
    }

    // sign_m_722
    private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                          boolean prioritized) 
    {
        Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);  // 查找统计节点
        if (selectedNode == null) {
            return true;
        }
        return rule.getRater().canPass(selectedNode, acquireCount, prioritized);    // 使用规则控制器进行判断，ref: sign_m_731
    }

com.alibaba.csp.sentinel.slots.block.flow.controller.DefaultController

    // sign_m_731 判断是否可以通过
    @Override
    public boolean canPass(Node node, int acquireCount, boolean prioritized) {
        int curCount = avgUsedTokens(node);     // sign_m_732
        if (curCount + acquireCount > count) {  // 判断是否超过自身设置的限制数
            if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {
                long currentTime = TimeUtil.currentTimeMillis();
                long waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);
                if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {
                    node.addWaitingRequest(currentTime + waitInMs, acquireCount);
                    node.addOccupiedPass(acquireCount);
                    sleep(waitInMs);
                    throw new PriorityWaitException(waitInMs);  // 报此异常可通过
                }
            }
            return false;   // 超过限制则返回 false (表示不通过)
        }
        return true;
    }

    // sign_m_732 返回当前计数
    private int avgUsedTokens(Node node) {
        if (node == null) {
            return DEFAULT_AVG_USED_TOKENS;
        }
        return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
    }

DegradeSlot

熔断处理
com.alibaba.csp.sentinel.slots.block.degrade.DegradeSlot

@Spi(order = Constants.ORDER_DEGRADE_SLOT)
public class DegradeSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable 
    {
        performChecking(context, resourceWrapper);                              // 先校验 sign_m_801
        fireEntry(context, resourceWrapper, node, count, prioritized, args);    // 再传给下游
    }

    // sign_m_801 熔断校验
    void performChecking(Context context, ResourceWrapper r) throws BlockException {
        List<CircuitBreaker> circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
        if (circuitBreakers == null || circuitBreakers.isEmpty()) {
            return;
        }
        for (CircuitBreaker cb : circuitBreakers) {
            if (!cb.tryPass(context)) { // 熔断判断 sign_m_810
                // 不通过则进行熔断报错
                throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule());
            }
        }
    }

    @Override // 退出时进行计数和状态处理
    public void exit(Context context, ResourceWrapper r, int count, Object... args) {
        Entry curEntry = context.getCurEntry();

        ... // 出现熔断 (有熔断异常) 则传给下游处理并返回

        List<CircuitBreaker> circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());

        ... // 无断路器则传给下游处理并返回

        if (curEntry.getBlockError() == null) {
            for (CircuitBreaker circuitBreaker : circuitBreakers) {
                circuitBreaker.onRequestComplete(context);  // 断路器计数与状态变更处理，ref: sign_m_830
            }
        }

        fireExit(context, r, count, args);
    }
}

com.alibaba.csp.sentinel.slots.block.degrade.circuitbreaker.AbstractCircuitBreaker

    // sign_m_810 熔断判断
    @Override
    public boolean tryPass(Context context) {
        if (currentState.get() == State.CLOSED) {
            return true;    // 断路器关闭：直接通过
        }
        if (currentState.get() == State.OPEN) {
            // 断路器已打开：超过指定熔断时间，尝试半打开处理
            return retryTimeoutArrived() && fromOpenToHalfOpen(context);    // sign_m_811 | sign_m_812
        }
        return false;       // 断路器半打开：不通过
    }

    // sign_m_811 判断是否超过熔断时长
    protected boolean retryTimeoutArrived() {
        return TimeUtil.currentTimeMillis() >= nextRetryTimestamp;
    }

    // sign_m_812 尝试半打开处理
    protected boolean fromOpenToHalfOpen(Context context) {
        if (currentState.compareAndSet(State.OPEN, State.HALF_OPEN)) {
            notifyObservers(State.OPEN, State.HALF_OPEN, null);
            Entry entry = context.getCurEntry();
            entry.whenTerminate(new BiConsumer<Context, Entry>() {  // 添加 entry.exit() 回调
                @Override
                public void accept(Context context, Entry entry) {  // 在 entry.exit() 被调用
                    if (entry.getBlockError() != null) {                            // 尝试请求时出错
                        currentState.compareAndSet(State.HALF_OPEN, State.OPEN);    // 重新打开
                        notifyObservers(State.HALF_OPEN, State.OPEN, 1.0d);
                    }
                }
            });
            return true;    // 只让一个线程 (且只进行一次) 处理
        }
        return false;       // 被其他线程抢占 (不通过)
    }

com.alibaba.csp.sentinel.slots.block.degrade.circuitbreaker.ExceptionCircuitBreaker

    // sign_m_830 断路器计数与状态变更处理
    @Override
    public void onRequestComplete(Context context) {
        Entry entry = context.getCurEntry();
        if (entry == null) {
            return;
        }
        Throwable error = entry.getError();
        SimpleErrorCounter counter = stat.currentWindow().value();
        if (error != null) {
            counter.getErrorCount().add(1); // 有异常，添加异常计数
        }
        counter.getTotalCount().add(1);     // 添加请求 (总) 计数

        handleStateChangeWhenThresholdExceeded(error);  // sign_m_831
    }

    // sign_m_831 状态变更处理
    private void handleStateChangeWhenThresholdExceeded(Throwable error) {
        ... // 当前状态为打开，则返回
        
        if (currentState.get() == State.HALF_OPEN) {// 当前为半打开状态
            if (error == null) {
                fromHalfOpenToClose();              // 无异常，则关闭
            } else {
                fromHalfOpenToOpen(1.0d);           // 有异常，则继续打开
            }
            return;
        }
        
        List<SimpleErrorCounter> counters = stat.values();
        long errCount = 0;
        long totalCount = 0;
        for (SimpleErrorCounter counter : counters) {
            errCount += counter.errorCount.sum();
            totalCount += counter.totalCount.sum();
        }
        if (totalCount < minRequestAmount) {
            return; // 小于最小请求数，不处理
        }
        double curCount = errCount;
        if (strategy == DEGRADE_GRADE_EXCEPTION_RATIO) {
            curCount = errCount * 1.0d / totalCount;    // 使用比率计算
        }
        if (curCount > threshold) {
            transformToOpen(curCount);                  // 超过阈值，开启熔断 (并设置开始重试的时间戳)
        }
    }