Sentinel之调用链分析

作者: 九点半的马拉 | 来源:发表于2020-04-22 22:40 被阅读0次

在该篇文章，将会分析是如何进入Sentinel开始执行操作的。大框架是基于构建一个调用链和处理链。

入口

下面是一个经典的Sentinel执行限流降级的代码。我们可以知道SphU.entry("HelloWorld")是执行逻辑代码的第一步。

public static void main(String[] args) {
    try {
        Entry entry = SphU.entry("HelloWorld");
        //业务代码
    } catch (BlockException ex) {
        // 处理被流控的逻辑
        System.out.println("blocked!");
    }catch (Exception e){
        e.printStackTrace();
    } finally {
        if (entry != null){
            entry.exit();
        }
    }
}

Entry

在Entry中，name是受保护资源的唯一姓名。
count设置为1，是通过请求要消耗的令牌数。

public static Entry entry(String name) throws BlockException {
        return Env.sph.entry(name, EntryType.OUT, 1, OBJECTS0);
}

然后会跳转到下面一步，根据资源名封装成一个StringResourceWrapper，一个资源包装类。

@Override
    public Entry entry(String name, EntryType type, int count, Object... args) throws BlockException {
        StringResourceWrapper resource = new StringResourceWrapper(name, type);
        return entry(resource, count, args);
}

public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
        return entryWithPriority(resourceWrapper, count, false, args);
}

下面是StringResourceWrapper的hashcode()和equals()方法，从中可以看出判断两个Resource是否相等，就是判断资源名是否相等。

@Override
public int hashCode() {
    return getName().hashCode();
}
@Override
public boolean equals(Object obj) {
    if (obj instanceof ResourceWrapper) {
        ResourceWrapper rw = (ResourceWrapper)obj;
        return rw.getName().equals(getName());
    }
    return false;
}

之后进入entryWithPriority进行设置上下文信息和寻找构建ProcessorSlot，这个方法很重要。

entryWithPriority

private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
        throws BlockException {
        # 获取当前上下文
        Context context = ContextUtil.getContext();
        if (context instanceof NullContext) {
            return new CtEntry(resourceWrapper, null, context);
        }
        if (context == null) {
            //获取默认的上下文 sentinel_default_context
            context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
        }

        //当全局开关关闭，则不会触发设定的规则
        if (!Constants.ON) {
            return new CtEntry(resourceWrapper, null, context);
        }
        //根据资源名寻找处理链
        ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
        //当resource超过6000个时，将不会创建新的chain,则不会执行限流，降级等操作
        if (chain == null) {
            return new CtEntry(resourceWrapper, null, context);
        }
        //为当前资源Resource和当前Context下创建一个CtEntry
        Entry e = new CtEntry(resourceWrapper, chain, context);
        try {
            //开始调用处理链
            chain.entry(context, resourceWrapper, null, count, prioritized, args);
        } catch (BlockException e1) {
            //出现异常，执行退出，并向上抛异常
            e.exit(count, args);
            throw e1;
        } catch (Throwable e1) {
            
        }
        return e;
}

上面是整体的流程，接下来分层进行解析

获取当前线程的上下文

下面是一个静态内部类，只有在调用的时候才会被加载，进行创建Context

 private final static class InternalContextUtil extends ContextUtil {
        static Context internalEnter(String name) {
            return trueEnter(name, "");
        }

        static Context internalEnter(String name, String origin) {
            return trueEnter(name, origin);
        }
}

下面是根据上下文名获取当前线程中的上下文，如果没有，则创建；本地有一个全局缓存的map，key为上下文名，value是EntranceNode，如果没有则新建一个EntranceNode，然后新建一个Context。

//key为上下文名，value是 EntranceNode
private static volatile Map<String, DefaultNode> contextNameNodeMap = new HashMap<>();

protected static Context trueEnter(String name, String origin) {
        Context context = contextHolder.get();
        if (context == null) {
            //key为上下文名，value为EntranceNode
            Map<String, DefaultNode> localCacheNameMap = contextNameNodeMap;
            DefaultNode node = localCacheNameMap.get(name);
            if (node == null) {
                //当上下文个数超过默认值2000后，将不再创建新的Context,返回一个NullContext
                if (localCacheNameMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {
                    setNullContext();
                    return NULL_CONTEXT;
                } else {
                    try {
                        LOCK.lock();
                        //上锁，双重验证
                        node = contextNameNodeMap.get(name);
                        if (node == null) {
                            if (contextNameNodeMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {
                                setNullContext();
                                return NULL_CONTEXT;
                            } else {
                                //创建一个入口节点，这里的name是上下文名
                                node = new EntranceNode(new StringResourceWrapper(name, EntryType.IN), null);
                                //所有上下文所处的树都有一个虚拟的根节点"machine-root",
                                Constants.ROOT.addChild(node);

                                Map<String, DefaultNode> newMap = new HashMap<>(contextNameNodeMap.size() + 1);
                                newMap.putAll(contextNameNodeMap);
                                newMap.put(name, node);
                                contextNameNodeMap = newMap;
                            }
                        }
                    } finally {
                        LOCK.unlock();
                    }
                }
            }  
            context = new Context(node, name);
            context.setOrigin(origin);
            //将context设置到当前Thread的threadlocalMap中
            contextHolder.set(context);
        }
        return context;
}

寻找当前资源的处理链

每一个Resource都有有个全局的ProcessorSlotChain，即在任何上下文中的同名资源共享一个ProcessorSlotChain。当Resource超过默认设置的6000个时，将不会构建ProcessorSlotChain，那么也不会执行后续的限流降级有关操作。

//相同的Resource将共享一个ProcessorSlotChain，不管在哪个上下文中
private static volatile Map<ResourceWrapper, ProcessorSlotChain> chainMap = new HashMap<ResourceWrapper, ProcessorSlotChain>();

ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
        ProcessorSlotChain chain = chainMap.get(resourceWrapper);
        if (chain == null) {
            synchronized (LOCK) {
                chain = chainMap.get(resourceWrapper);
                if (chain == null) {
                    // 当超过6000个后，则不创建
                    if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
                        return null;
                    }
                    //加载一个chain
                    chain = SlotChainProvider.newSlotChain();
                    Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
                        chainMap.size() + 1);
                    newMap.putAll(chainMap);
                    newMap.put(resourceWrapper, chain);
                    chainMap = newMap;
                }
            }
        }
        return chain;
}

下面是创建chain的方法，可以通过SPI机制进行加载，我选择了自带的构建方法DefaultSlotChainBuilder来分析。

public static ProcessorSlotChain newSlotChain() {
        if (slotChainBuilder != null) {
            return slotChainBuilder.build();
        }
        // Resolve the slot chain builder SPI.
        slotChainBuilder = SpiLoader.loadFirstInstanceOrDefault(SlotChainBuilder.class, DefaultSlotChainBuilder.class);

        if (slotChainBuilder == null) {
            slotChainBuilder = new DefaultSlotChainBuilder();
        } else {
            RecordLog.info("[SlotChainProvider] Global slot chain builder resolved: "
                + slotChainBuilder.getClass().getCanonicalName());
        }
        return slotChainBuilder.build();
    }

    private SlotChainProvider() {}
}

最新版本的DefaultSlotChainBuilder也改用SPI机制了，原来的不是。

public class DefaultSlotChainBuilder implements SlotChainBuilder {

    @Override
    public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();
        List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);
        for (ProcessorSlot slot : sortedSlotList) {
            if (!(slot instanceof AbstractLinkedProcessorSlot)) {
                continue;
            }
            chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
        }
        return chain;
    }
}

下面是旧版本，

public class DefaultSlotChainBuilder implements SlotChainBuilder {
    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 AuthoritySlot());
        chain.addLast(new SystemSlot());
        chain.addLast(new FlowSlot());
        chain.addLast(new DegradeSlot());
        return chain;
    }
}

创建一个CtEntry

CtEntry在文档中解释为Linked entry within current context，
而它是继承Entry，在文档中解释为This class holds information of current invocation。

一个上下文中会有一个调用树invocation tree，就是利用CtEntry实现的，它有parent和child变量。

class CtEntry extends Entry {
    protected Entry parent = null;
    protected Entry child = null;

    protected ProcessorSlot<Object> chain;
    protected Context context;

在创建一个Entry实例时，则会把相关的额Resource、chain和context赋值给他，最后的setUpEntryFor方法很关键。

CtEntry(ResourceWrapper resourceWrapper, ProcessorSlot<Object> chain, Context context) {
        super(resourceWrapper);
        this.chain = chain;
        this.context = context;
        setUpEntryFor(context);
    }

从下面可以得知在同一个调用树中，可以有多个Entry，因为我们在调用一个服务的时候可能在中间也需要再调用其他的服务，所以会存在多个Entry。

在创建一个Entry的时候，需要将该Entry设置文当前Context的当前的Entry，因为将要处理该Entry。

private void setUpEntryFor(Context context) {
        // The entry should not be associated to NullContext.
        if (context instanceof NullContext) {
            return;
        }
        //将之前context所记录的当前Entry设置为新建Entry的父节点
        this.parent = context.getCurEntry();
        if (parent != null) {
            //如果父节点不为空，则设置父节点的子节点为新建的Entry
            ((CtEntry)parent).child = this;
        }
        //最后将新建的Entry设置为所处Context的当前Entry。
        context.setCurEntry(this);
}

chain.entry 调用处理链

它主要是执行每个Slot中的entry()方法。

我将具体的讲解每个Slot的处理方法。

NodeSelectorSlot

首先是调用NodeSelectorSlot的entry()方法。它的职责就是在当前上下文下，为当前Entry添加对应的DefaultNode，并设置Context的当前Entry中的当前节点为该节点DefaultNode。

下面有一个map。使用时context名作为key,而不是资源名。因为在不同Context下相同资源共享同一个ProcessorSlotChain，所以可能是在不同的Context下调用该entry方法，利用Context作为key可以区分不同Context下的数据统计。

简单的说，就是所有上下文的同一资源对应一个ProcessorSlotChain,一个ProcessorSlotChain有一个map,即所有上下文的同一资源拥有同一个map,要想获取在每个上下文的该资源的调用情况，所以key为上下文名，

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) {
                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;
                    // Build invocation tree
                    ((DefaultNode) context.getLastNode()).addChild(node);
                }
            }
        }
        //设置新建的node为当前上下文的当前节点
        context.setCurNode(node);
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
    }

下面是获取当前节点的父节点，此时的CurEntry就是新建的Entry，从上面可知，在新建Entry时，会将上一个Entry设置为它的父节点。

public Node getLastNode() {
        if (curEntry != null && curEntry.getLastNode() != null) {
            return curEntry.getLastNode();
        } else {
            return entranceNode;
        }
}

CtEntry类
@Override
public Node getLastNode() {
       return parent == null ? null : parent.getCurNode();
}

从上面可知，会先获取当前新建Entry的父节点，然后将新建Entry所对应的DefalutNode加入到父节点的ChildList集合中，以Node的维度记录他的子节点

private volatile Set<Node> childList = new HashSet<>();

public void addChild(Node node) {
        if (node == null) {
            return;
        }
        if (!childList.contains(node)) {
            synchronized (this) {
                if (!childList.contains(node)) {
                    Set<Node> newSet = new HashSet<>(childList.size() + 1);
                    newSet.addAll(childList);
                    newSet.add(node);
                    childList = newSet;
                }
            }
        }
}

从下面可以看出，它会调用下一个Slot的entry方法。

@Override
public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
        throws Throwable {
        if (next != null) {
            next.transformEntry(context, resourceWrapper, obj, count, prioritized, args);
        }
    }
    
void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args)
        throws Throwable {
        T t = (T)o;
        entry(context, resourceWrapper, t, count, prioritized, args);
}

ClusterBuilderSlot

为了统计相同资源在所有的上下文中的总的数据，我们采用clusterNode来统计。不同上下文相同资源共享同一个clusterNode。

该Slot的主要任务就是为相同资源共享一个ClusterNode，并缓存起来，并将当前DefalutNode的clusterNode指向它。

private static volatile Map<ResourceWrapper, ClusterNode> clusterNodeMap = new HashMap<>();  

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

LogSlot

先执行后续的Slot，如果有异常，就会将设置的block的信息记录到日志文件sentinel-block.log中。

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

}

StatisticSlot

该slot在前面的一篇文章讲述过了，主要是做数据的实时统计。

SystemSlot

Sentinel针对所有的入口流量，使用一个全局的ENTRY_NODE进行统计，系统保护规则是全局的，和具体的某个资源没有关系。

AuthoritySlot

主要是根据黑白名单进行过滤。

FlowSlot

FlowSlot主要是根据前面统计好的信息，与设置的限流规则进行匹配校验，如果规则校验不通过，则进行限流。

DegradeSlot

DegradeSlot 主要是根据前面统计好的信息，与设置的降级规则进行匹配校验，如果规则校验不通过，则进行降级。

后面的几个Slot将在以后的文章进行撰写。

参考文章：
Sentinel 原理-调用链
 Sentinel 调用上下文环境实现原理(含原理图)
寻找一把进入 Alibaba Sentinel 的钥匙(文末附流程图)

Sentinel之调用链分析

入口

Entry

entryWithPriority

获取当前线程的上下文

寻找当前资源的处理链

创建一个CtEntry

chain.entry 调用处理链

NodeSelectorSlot

ClusterBuilderSlot

LogSlot

StatisticSlot

SystemSlot

AuthoritySlot

FlowSlot

DegradeSlot

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

服务监控和治理