sentinal源码3－资源处理链

2019-03-30 本文已影响0人 modou1618

一处理链工厂函数

1.1 SlogChainProvider

静态初始化获取处理链工厂类
static final ServiceLoader<SlotChainBuilder> LOADER = ServiceLoader.load(SlotChainBuilder.class);
可自定义处理链工厂，或使用默认工厂
DefaultSlotChainBuilder默认工厂创建资源处理链

public ProcessorSlotChain build() {
    ProcessorSlotChain chain = new DefaultProcessorSlotChain();
    chain.addLast(new NodeSelectorSlot());
    chain.addLast(new ClusterBuilderSlot());
    chain.addLast(new LogSlot());
    chain.addLast(new StatisticSlot());
    chain.addLast(new SystemSlot());
    chain.addLast(new AuthoritySlot());
    chain.addLast(new FlowSlot());
    chain.addLast(new DegradeSlot());

    return chain;
}

二处理链头节点

2.1 类关系

image.png

2.2 AbstractLinkedProcessorSlot

所有处理链节点都继承AbstractLinkedProcessorSlot
AbstractLinkedProcessorSlot内部有一个next属性，用于维持一个资源处理函数的单链表。
触发下一节点的entry函数调用
void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
触发下一节点的exit函数调用
void fireExit(Context context, ResourceWrapper resourceWrapper, int count, Object... args);

2.3 DefaultProcessorSlotChain

提供链表节点前插入和后插入的接口
提供entry和exit函数调用接口，依次按顺序调用单链表上各节点的对应函数

三处理链节点

每个资源一个处理链，即使在不同context上下文中

3.1 NodeSelectorSlot()

对于每个资源，为当前context上下文创建一个DefaultNode统计节点
记录节点父子关系，记录节点为context当前处理节点，即对当前请求资源进行统计。

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) {
            node = map.get(context.getName());
            if (node == null) {
                node = Env.nodeBuilder.buildTreeNode(resourceWrapper, null);
                HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());
                cacheMap.putAll(map);
                cacheMap.put(context.getName(), node);
                map = cacheMap;
            }
            // Build invocation tree
            ((DefaultNode)context.getLastNode()).addChild(node);
        }
    }

    context.setCurNode(node);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

3.2 ClusterBuilderSlot()

按资源名静态存储资源统计ClusterNode节点，即每个资源只创建一个对应ClusterNode
static volatile Map<ResourceWrapper, ClusterNode> clusterNodeMap = new HashMap<>();
获取资源对应的ClusterNode，存储到当前context下资源的DefaultNode.clusterNode属性中
若指定了资源访问来源origin，则对资源的每个来源只创建一个StatisticNode统计该来源的资源请求数据。所有来源的统计节点存储在ClusterNode.originCountMap中。
Map<String, StatisticNode> originCountMap

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args)
    throws Throwable {
    if (clusterNode == null) {
        synchronized (lock) {
            if (clusterNode == null) {
                // Create the cluster node.
                clusterNode = Env.nodeBuilder.buildClusterNode();
                HashMap<ResourceWrapper, ClusterNode> newMap = new HashMap<>(Math.max(clusterNodeMap.size(), 16));
                newMap.putAll(clusterNodeMap);
                newMap.put(node.getId(), clusterNode);

                clusterNodeMap = newMap;
            }
        }
    }
    node.setClusterNode(clusterNode);

    if (!"".equals(context.getOrigin())) {
        Node originNode = node.getClusterNode().getOrCreateOriginNode(context.getOrigin());
        context.getCurEntry().setOriginNode(originNode);
    }

    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

3.3 LogSlot()

在资源请求时，记录请求失败日志，和请求异常日志

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(), count);
        throw e;
    } catch (Throwable e) {
        RecordLog.warn("Unexpected entry exception", e);
    }

}

3.4 StatisticSlot()

3.4.1 请求时统计

先做后续资源请求判断，成功则添加成功统计，失败则添加失败统计

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args) throws Throwable {
    try {
        // 先进行后续处理
        fireEntry(context, resourceWrapper, node, count, prioritized, args);

        // 资源请求成功，添加资源线程数，和资源使用数统计
//同时添加当前context上下文下的DefaultNode中的资源统计和资源唯一节点ClusterNode中的资源统计
        node.increaseThreadNum();
        node.addPassRequest(count);

        if (context.getCurEntry().getOriginNode() != null) {
            //根据调用来源添加资源请求统计数据
            context.getCurEntry().getOriginNode().increaseThreadNum();
            context.getCurEntry().getOriginNode().addPassRequest(count);
        }

        if (resourceWrapper.getType() == EntryType.IN) {
            // 入口资源的统计增加
            Constants.ENTRY_NODE.increaseThreadNum();
            Constants.ENTRY_NODE.addPassRequest(count);
        }

        //资源请求成功后的回调函数调用
        for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
            handler.onPass(context, resourceWrapper, node, count, args);
        }
    } catch (PriorityWaitException ex) {
        node.increaseThreadNum();
        if (context.getCurEntry().getOriginNode() != null) {
            // Add count for origin node.
            context.getCurEntry().getOriginNode().increaseThreadNum();
        }

        if (resourceWrapper.getType() == EntryType.IN) {
            // Add count for global inbound entry node for global statistics.
            Constants.ENTRY_NODE.increaseThreadNum();
        }
        // Handle pass event with registered entry callback handlers.
        for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
            handler.onPass(context, resourceWrapper, node, count, args);
        }
    } catch (BlockException e) {
        // Blocked, set block exception to current entry.
        context.getCurEntry().setError(e);

        // Add block count.
        node.increaseBlockQps(count);
        if (context.getCurEntry().getOriginNode() != null) {
            context.getCurEntry().getOriginNode().increaseBlockQps(count);
        }

        if (resourceWrapper.getType() == EntryType.IN) {
            // Add count for global inbound entry node for global statistics.
            Constants.ENTRY_NODE.increaseBlockQps(count);
        }

        // Handle block event with registered entry callback handlers.
        for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
            handler.onBlocked(e, context, resourceWrapper, node, count, args);
        }

        throw e;
    } catch (Throwable e) {
        // Unexpected error, set error to current entry.
        context.getCurEntry().setError(e);

        // This should not happen.
        node.increaseExceptionQps(count);
        if (context.getCurEntry().getOriginNode() != null) {
            context.getCurEntry().getOriginNode().increaseExceptionQps(count);
        }

        if (resourceWrapper.getType() == EntryType.IN) {
            Constants.ENTRY_NODE.increaseExceptionQps(count);
        }
        throw e;
    }
}

3.4.2 释放资源时统计数据处理

请求成功则做一些统计数据处理
资源线程数-1，
记录资源持有时间，
存在来源配置则也记录指定来源的资源持有时间
入口类型，则更新入口节点统计
exit回调函数调用

public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
    DefaultNode node = (DefaultNode)context.getCurNode();

    if (context.getCurEntry().getError() == null) {
        // Calculate response time (max RT is TIME_DROP_VALVE).
        long rt = TimeUtil.currentTimeMillis() - context.getCurEntry().getCreateTime();
        if (rt > Constants.TIME_DROP_VALVE) {
            rt = Constants.TIME_DROP_VALVE;
        }

        // Record response time and success count.
        node.addRtAndSuccess(rt, count);
        if (context.getCurEntry().getOriginNode() != null) {
            context.getCurEntry().getOriginNode().addRtAndSuccess(rt, count);
        }

        node.decreaseThreadNum();

        if (context.getCurEntry().getOriginNode() != null) {
            context.getCurEntry().getOriginNode().decreaseThreadNum();
        }

        if (resourceWrapper.getType() == EntryType.IN) {
            Constants.ENTRY_NODE.addRtAndSuccess(rt, count);
            Constants.ENTRY_NODE.decreaseThreadNum();
        }
    } else {
        // Error may happen.
    }

    // Handle exit event with registered exit callback handlers.
    Collection<ProcessorSlotExitCallback> exitCallbacks = StatisticSlotCallbackRegistry.getExitCallbacks();
    for (ProcessorSlotExitCallback handler : exitCallbacks) {
        handler.onExit(context, resourceWrapper, count, args);
    }

    fireExit(context, resourceWrapper, count);
}

3.5 SystemSlot()

系统级资源限流规则匹配，即EntryType.IN类型资源

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args) throws Throwable {
    SystemRuleManager.checkSystem(resourceWrapper);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

3.6 AuthoritySlot()

授权规则匹配

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)
    throws Throwable {
    checkBlackWhiteAuthority(resourceWrapper, context);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

3.7 FlowSlot()

流控规则匹配

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);
}

3.8 DegradeSlot()

降级规则匹配

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)
    throws Throwable {
    DegradeRuleManager.checkDegrade(resourceWrapper, context, node, count);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

sentinal源码3－资源处理链

一处理链工厂函数

1.1 SlogChainProvider

二处理链头节点

2.1 类关系

2.2 AbstractLinkedProcessorSlot

2.3 DefaultProcessorSlotChain

三处理链节点

3.1 NodeSelectorSlot()

3.2 ClusterBuilderSlot()

3.3 LogSlot()

3.4 StatisticSlot()

3.4.1 请求时统计

3.4.2 释放资源时统计数据处理

3.5 SystemSlot()

3.6 AuthoritySlot()

3.7 FlowSlot()

3.8 DegradeSlot()

猜你喜欢

热点阅读

sentinal源码3－资源处理链

一 处理链工厂函数

1.1 SlogChainProvider

二 处理链头节点

2.1 类关系

2.2 AbstractLinkedProcessorSlot

2.3 DefaultProcessorSlotChain

三 处理链节点

3.1 NodeSelectorSlot()

3.2 ClusterBuilderSlot()

3.3 LogSlot()

3.4 StatisticSlot()

3.4.1 请求时统计

3.4.2 释放资源时统计数据处理

3.5 SystemSlot()

3.6 AuthoritySlot()

3.7 FlowSlot()

3.8 DegradeSlot()

猜你喜欢

热点阅读

一处理链工厂函数

二处理链头节点

三处理链节点