Alibaba

admin管理员组

文章数量:1122854

Alibaba

Alibaba-Sentinel

两种熔断降级框架的对比

Sentinel的总体流程

在Sentinel中，如果要对模块一块代码进行限流、熔断等，需要定义一个资源，然后将要限流、熔断的代码块包裹起来。

在 Sentinel 里面，所有的资源都对应一个资源名称（resourceName），每次资源调用都会创建一个 Entry 对象。Entry 可以通过对主流框架的适配自动创建，也可以通过注解的方式或调用 SphU API 显式创建。Entry 创建的时候，同时也会创建一系列功能插槽（slot chain），以使用SphU.entry(String name)来获取资源的例子进行分析

// SphU的entry方法 实际上是调用了CtSph的entry方法
public static Entry entry(String name) throws BlockException {return Env.sph.entry(name, EntryType.OUT, 1, OBJECTS0);
}

最终调用了CtSph的entryWithPriority方法，由这里我们可以看出，每次调用SphU.entry都会创建一个Entry入口。

private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)throws BlockException {// 1.从contextHolder(基于ThreadLocal的实现)获取contextContext context = ContextUtil.getContext();if (context instanceof NullContext) {// The {@link NullContext} indicates that the amount of context has exceeded the threshold,// so here init the entry only. No rule checking will be done.return new CtEntry(resourceWrapper, null, context);}if (context == null) {// 2.这个方法会创建获取一个默认context，意味着如果不指定context，多线程时所有的线程获取的也是这个默            // 认context,这个默认context叫Constants.CONTEXT_DEFAULT_NAMEcontext = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);}// Global switch is close, no rule checking will do.if (!Constants.ON) {return new CtEntry(resourceWrapper, null, context);}// 3.寻找slotchain，slotchain就是利用责任链模式实现sentinel功能的核心ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);/** Means amount of resources (slot chain) exceeds {@link Constants.MAX_SLOT_CHAIN_SIZE},* so no rule checking will be done.*/if (chain == null) {return new CtEntry(resourceWrapper, null, context);}// 创建了Entry入口，每次调用SphU都会创建Entry入口，并把entry设置到context里// 后续再说context里的内容及其数据结构Entry e = new CtEntry(resourceWrapper, chain, context);try {// 开始进入了责任链，进行sentinel的功能chain.entry(context, resourceWrapper, null, count, prioritized, args);} catch (BlockException e1) {e.exit(count, args);throw e1;} catch (Throwable e1) {// This should not happen, unless there are errors existing in Sentinel internal.RecordLog.info("Sentinel unexpected exception", e1);}return e;}

上面的方法构造了一个如下的初始化Context，记住每个线程获取资源前没有context时都会new一个Context，是不同的对象，但是他们的contextName也许是一样。entranceNode是和contextName关联的，所以同一个contextName只会有一个entranceNode，从后续的分析可以知道，entranceNode下挂着一个DefaultNode，这个DefaultNode也是和contextName对应的，而且DefaultNode是继承自StatisticNode,它里面可以保存一些流量数据，因为contextName和defaultNode是一对一关系，所以可以用给context设置流控规则，其实就是利用的EntranceNode下的defaultNode。而给每个资源设置的流控规则，则是这个clusterNode。详细可见后面的各种slot。

创建过滤链的代码

ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {// ResourceWrapper的hashCode和equals方法都是根据name重写的，所以同一个资源名称就有相同的chainProcessorSlotChain chain = chainMap.get(resourceWrapper);if (chain == null) {synchronized (LOCK) {chain = chainMap.get(resourceWrapper);if (chain == null) {// Entry size limit.if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {return null;}// 这个方法构建了chainchain = SlotChainProvider.newSlotChain();Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(chainMap.size() + 1);newMap.putAll(chainMap);newMap.put(resourceWrapper, chain);chainMap = newMap;}}}return chain;}

这些插槽有不同的职责，例如:

• NodeSelectorSlot 负责收集资源的路径，并将这些资源的调用路径，以树状结构存储起来，用于根据调用路径来限流降级；
• ClusterBuilderSlot 则用于存储资源的统计信息以及调用者信息，例如该资源的 RT, QPS, thread count 等等，这些信息将用作为多维度限流，降级的依据；
• StatisticSlot 则用于记录、统计不同纬度的 runtime 指标监控信息；
• FlowSlot 则用于根据预设的限流规则以及前面 slot 统计的状态，来进行流量控制；
• AuthoritySlot 则根据配置的黑白名单和调用来源信息，来做黑白名单控制；
• DegradeSlot 则通过统计信息以及预设的规则，来做熔断降级；
• SystemSlot 则通过系统的状态，例如 load1 等，来控制总的入口流量；

通过slot chain，实现了Sentinel的限流、熔断、降级、系统保护等功能。

NodeSelectorSlot

NodeSelectorSlot 是用来构造调用链的，具体的是将资源(Resource,Sentinel的原理是进入你要保护的代码或接口前要获取一下资源，以资源获取的qps为限流标准)的调用路径，封装成一个一个的节点，再组成一个树状的结构来形成一个完整的调用链。

@Overridepublic void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)throws Throwable {// 通一个context对应的DefaultNode是一样的，先根据contextName获取DefaultNodeDefaultNode 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;// 构造调用树((DefaultNode) context.getLastNode()).addChild(node);}}}context.setCurNode(node);fireEntry(context, resourceWrapper, node, count, prioritized, args);}

当不同Context进入相同的资源时，DefaultNode的是不相同的，Entry是不同的。DefaultNode也可以存储一个上下文的qps，用于流控。

相同Context下进入不同的资源时如下,即在同一Context下调用SphU.entry(resourceName)，进入不同的资源

ClusterBuilderSlot

ClusterBuilderSlot的作用就是根据资源找到或创建ClusterNode，并设置到DefaultNode下。ClusteNode是每个资源都对应唯一一个，同一资源的所有统计数据都是在ClusteNode里。因为DefaultNode在同一Context中是相同的，所以同一个Context中，访问不同的资源，在此Slot里面会将DefaultNode的ClusteNode属性改变。

StatisticSlot

这个Slot用于统计资源的qps，主要是对上述的ClusterNode进行操作，ClusterNode又是继承自StatisticNode。看一下StatisticNode.addPassRequest方法。

public void addPassRequest(int count) {rollingCounterInSecond.addPass(count);rollingCounterInMinute.addPass(count);
}

有两种刻度，分别是秒和分钟，这里的统计方法是利用滑动窗口模式进行统计的。

/**  通过点击追踪构造函数，发现其是一个LeapArray的包装类，看了LeapArray的构造方法，得出这几个参数的含义* Holds statistics of the recent {@code INTERVAL} seconds. The {@code INTERVAL} is divided into time spans* by given {@code sampleCount}.sampleCount代表窗口数量，INTERVAL代表总时间，也就是这个数组能保留INTERVAL时间的流量数据，且按INTERVAL / sampleCount的个窗口划分该时段的流量数据。*/private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,IntervalProperty.INTERVAL);/*** Holds statistics of the recent 60 seconds. The windowLengthInMs is deliberately set to 1000 milliseconds,* meaning each bucket per second, in this way we can get accurate statistics of each second.60是窗口数， 60 * 1000是数组能保留的总时间， 60 * 1000 / 60 则是窗口的大小即窗口大小为一分钟*/private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);

LeapArray是一个滑动窗口的实现类，可以用它获取一个当前窗口。下面分析一些它的滑动窗口代码

public WindowWrap<T> currentWindow(long timeMillis) {if (timeMillis < 0) {return null;}// 算出当前时间对应窗口数组的哪个位置，用当前时间除以窗口大小，再向数组大小取模可以得到int idx = calculateTimeIdx(timeMillis);// 计算出当前窗口的起始时间long windowStart = calculateWindowStart(timeMillis);/** Get bucket item at given time from the array.** (1) Bucket is absent, then just create a new bucket and CAS update to circular array.* (2) Bucket is up-to-date, then just return the bucket.* (3) Bucket is deprecated, then reset current bucket and clean all deprecated buckets.*/while (true) {WindowWrap<T> old = array.get(idx);if (old == null) {// 如果窗口不存在则创建窗口/**     B0       B1      B2    NULL      B4* ||_______|_______|_______|_______|_______||___* 200     400     600     800     1000    1200  timestamp*                             ^*                          time=888*            bucket is empty, so create new and update** If the old bucket is absent, then we create a new bucket at {@code windowStart},* then try to update circular array via a CAS operation. Only one thread can* succeed to update, while other threads yield its time slice.*/WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));if (arraypareAndSet(idx, null, window)) {// CAS保证线程安全// Successfully updated, return the created bucket.return window;} else {// Contention failed, the thread will yield its time slice to wait for bucket available.Thread.yield();}} else if (windowStart == old.windowStart()) {// 还是旧窗口，直接返回/**     B0       B1      B2     B3      B4* ||_______|_______|_______|_______|_______||___* 200     400     600     800     1000    1200  timestamp*                             ^*                          time=888*            startTime of Bucket 3: 800, so it's up-to-date** If current {@code windowStart} is equal to the start timestamp of old bucket,* that means the time is within the bucket, so directly return the bucket.*/return old;} else if (windowStart > old.windowStart()) {// 需要覆盖旧窗口/**   (old)*             B0       B1      B2    NULL      B4* |_______||_______|_______|_______|_______|_______||___* ...    1200     1400    1600    1800    2000    2200  timestamp*                              ^*                           time=1676*          startTime of Bucket 2: 400, deprecated, should be reset** If the start timestamp of old bucket is behind provided time, that means* the bucket is deprecated. We have to reset the bucket to current {@code windowStart}.* Note that the reset and clean-up operations are hard to be atomic,* so we need a update lock to guarantee the correctness of bucket update.** The update lock is conditional (tiny scope) and will take effect only when* bucket is deprecated, so in most cases it won't lead to performance loss.*/if (updateLock.tryLock()) {// 窗口更新是竞争操作，需要加锁try {// Successfully get the update lock, now we reset the bucket.return resetWindowTo(old, windowStart);} finally {updateLock.unlock();}} else {// Contention failed, the thread will yield its time slice to wait for bucket available.Thread.yield();}} else if (windowStart < old.windowStart()) {// Should not go through here, as the provided time is already behind.return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));}}}

整个获取滑动窗口的代码还是比较好理解的，分为上面的三种情况，还有一种在代码健壮时不可能发生的情况。

Sentinel与Feign结合使用的原理

SentinelFeignAutoConfiguration自动配置类会在feign.sentinel.enabled为true时创建动态代理。
SentinelInvocationHandler 是代理类，下面看部分源码。

本文标签： Alibaba