sentinal源码3－资源处理链

作者: modou1618 | 来源:发表于2019-03-30 14:32 被阅读0次

sentinal源码3－资源处理链
sentinal
sentinal源码1－入口
sentinal源码4－demo-basic处理流程
springcloud整合nacos、sentinal、spri
缓存架构之15：对项目中的哨兵节点进行管理以及高可用redis集
redis哨兵管理和高可用容灾演练
quartz源码3－调度处理
Spring 源码（二）注册内置的BeanPostProcess
.netcore高级工程师面试之Redis的哨兵架构

一处理链工厂函数

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 静态初始化获取处理链工厂类static final...
sentinal
sentinal源码：https://github.com/alibaba/Sentinel Sentinel 是...
sentinal源码1－入口
sentinal是阿里开源的限流降级的一个基础组件，本系列主要是记录本人对sentinal源码的理解。sentin...
sentinal源码4－demo-basic处理流程
sentinal提供了很多场景的demo程序，这里看下sentinel-demo-basic里面的几个demo 配...
springcloud整合nacos、sentinal、spri
源码地址:下载地址使用该架构的项目地址:下载地址下面教大家整合nacos、sentinal、springclou...
缓存架构之15：对项目中的哨兵节点进行管理以及高可用redis集
1、哨兵节点的增加和删除增加sentinal，会自动发现删除sentinal的步骤（1）停止sentinal进...
redis哨兵管理和高可用容灾演练
1、哨兵节点的增加和删除增加sentinal，会自动发现删除sentinal的步骤（1）停止sentinal...
quartz源码3－调度处理
一类依赖结构二调度线程初始化阶段等待其他流程初始化完成，进入调度处理数据库访问失败则睡眠后重试 Obje...
Spring 源码（二）注册内置的BeanPostProcess
发起注册调用链路：源码：注册Bean定义到容器调用链路：源码：这里需要注意的两个后置处理器分别是： A...
.netcore高级工程师面试之Redis的哨兵架构
简介：Github 1 Redis Sentinal机制 sentinal，中文名哨兵哨兵是redis集群架构中...

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－资源处理链

sentinal

sentinal源码1－入口

sentinal源码4－demo-basic处理流程

springcloud整合nacos、sentinal、spri

缓存架构之15：对项目中的哨兵节点进行管理以及高可用redis集

redis哨兵管理和高可用容灾演练

quartz源码3－调度处理

Spring 源码（二）注册内置的BeanPostProcess

.netcore高级工程师面试之Redis的哨兵架构

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

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()

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

一处理链工厂函数

二处理链头节点

三处理链节点