源码分析(spring alibaba 源码分析之 sentinel)
前两篇文章只画了流程图，发现反响不够好，没几个阅读量，今天我把看源码的方式也记录下，详细解析下主线逻辑。
sentinel实现其实就是一个大的Try catch ，在StatisticSlot保存各种指标，在其他slot中进行判断，通过就继续执行其他slot，不能通过就抛出异常，达到阈值就进行限流或者降级操作。
先上一个源码分析的流程图
1. sentinel是通过aop实现的它的注解是@SentinelResource那我们就找他的实现
2. 找到SentinelResourceAspect类的invokeResourceWithSentinel方法就是具体实现。
public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {
Method originMethod = this.resolveMethod(pjp);
SentinelResource annotation = (SentinelResource)originMethod.getAnnotation(SentinelResource.class);
if (annotation == null) {
throw new IllegalStateException("Wrong state for SentinelResource annotation");
} else {
String resourceName = this.getResourceName(annotation.value(), originMethod);
EntryType entryType = annotation.entryType();
int resourceType = annotation.resourceType();
Entry entry = null;
try {
Object var18;
try {
entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());
Object result = pjp.proceed();
var18 = result;
return var18;
} catch (BlockException var15) {
var18 = this.handleBlockException(pjp, annotation, var15);
return var18;
} catch (Throwable var16) {
Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();
if (exceptionsToIgnore.length > 0 && this.exceptionBelongsTo(var16, exceptionsToIgnore)) {
throw var16;
} else if (this.exceptionBelongsTo(var16, annotation.exceptionsToTrace())) {
this.traceException(var16, annotation);
Object var10 = this.handleFallback(pjp, annotation, var16);
return var10;
} else {
throw var16;
}
}
} finally {
if (entry != null) {
entry.exit(1, pjp.getArgs());
}
}
}
}
3 调用SphU.entry方法，一步步跟到具体实现，找到CtSph类的entryWithPriority方法。
获取slot的方法。
ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
ProcessorSlotChain 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;
}
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;
}
public static ProcessorSlotChain newSlotChain() {
if (builder != null) {
return builder.build();
}
resolveSlotChainBuilder();
if (builder == null) {
RecordLog.warn("[SlotChainProvider] Wrong state when resolving slot chain builder, using default");
builder = new DefaultSlotChainBuilder();
}
return builder.build();
}
public class DefaultSlotChainBuilder implements SlotChainBuilder {
@Override
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;
}
}
以上逻辑新建了8个slot，以及一个链条，这个就是sentinel核心实现。
回到entryWithPriority继续向下走。
try {
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);
}
执行
DefaultProcessorSlotChain.entry方法。
执行AbstractLinkedProcessorSlot.fireEntry方法
执行transformEntry方法。
transformEntry方法里面有entry，此时会一次执行(slot上面已经new出来了)slot的entry方法，如果在配置的各种指标内，会执行完成，如果大于配置的指标数据，会抛出对应异常。
实现流控/熔断/降级的就三个slot,我们就着重看着三个slot。
StatisticSlot是个统计指标的实现类，这里创建了sentinel的滑动时间窗口，我们来看看他是怎么实现的。
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
Iterator var8;
ProcessorSlotEntryCallback handler;
try {
this.fireEntry(context, resourceWrapper, node, count, prioritized, args);
node.increaseThreadNum();
node.addPassRequest(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseThreadNum();
context.getCurEntry().getOriginNode().addPassRequest(count);
}
if (resourceWrapper.getEntryType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseThreadNum();
Constants.ENTRY_NODE.addPassRequest(count);
}
Iterator var13 = StatisticSlotCallbackRegistry.getEntryCallbacks().iterator();
while(var13.hasNext()) {
ProcessorSlotEntryCallback<DefaultNode> handler = (ProcessorSlotEntryCallback)var13.next();
handler.onPass(context, resourceWrapper, node, count, args);
}
} catch (PriorityWaitException var10) {
node.increaseThreadNum();
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseThreadNum();
}
if (resourceWrapper.getEntryType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseThreadNum();
}
var8 = StatisticSlotCallbackRegistry.getEntryCallbacks().iterator();
while(var8.hasNext()) {
handler = (ProcessorSlotEntryCallback)var8.next();
handler.onPass(context, resourceWrapper, node, count, args);
}
} catch (BlockException var11) {
BlockException e = var11;
context.getCurEntry().setError(var11);
node.increaseBlockQps(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseBlockQps(count);
}
if (resourceWrapper.getEntryType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseBlockQps(count);
}
var8 = StatisticSlotCallbackRegistry.getEntryCallbacks().iterator();
while(var8.hasNext()) {
handler = (ProcessorSlotEntryCallback)var8.next();
handler.onBlocked(e, context, resourceWrapper, node, count, args);
}
throw e;
} catch (Throwable var12) {
context.getCurEntry().setError(var12);
node.increaseExceptionQps(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseExceptionQps(count);
}
if (resourceWrapper.getEntryType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseExceptionQps(count);
}
throw var12;
}
}
从代码可以看出，首先执行了this.fireEntry方法向下执行，如果执行成功对各种指标（执行成功数，执行线程数做++），如果执行失败对异常指标++；
这里面需要插入一段滑动时间窗口的实现。
在StatisticNode类中新建了一个rollingCounterInSecond
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
IntervalProperty.INTERVAL);
SampleCountProperty.SAMPLE_COUNT 默认2
IntervalProperty.INTERVAL默认1000
我们去看看他的具体实现
public LeapArray(int sampleCount, int intervalInMs) {
AssertUtil.isTrue(sampleCount > 0, "bucket count is invalid: " + sampleCount);
AssertUtil.isTrue(intervalInMs > 0, "total time interval of the sliding window should be positive");
AssertUtil.isTrue(intervalInMs % sampleCount == 0, "time span needs to be evenly divided");
this.windowLengthInMs = intervalInMs / sampleCount;
this.intervalInMs = intervalInMs;
this.sampleCount = sampleCount;
this.array = new AtomicReferenceArray<>(sampleCount);
}
这段代码的意思就是创建了两个窗口。
node.addPassRequest(count);
this.clusterNode.addPassRequest(count);
@Override
public void addPassRequest(int count) {
rollingCounterInSecond.addPass(count);
rollingCounterInMinute.addPass(count);
}
@Override
public void addPass(int count) {
WindowWrap<MetricBucket> wrap = data.currentWindow();
wrap.value().addPass(count);
}
这个就是获取时间窗口的方法
/**
* Get bucket item at provided timestamp.
*
* @param timeMillis a valid timestamp in milliseconds
* @return current bucket item at provided timestamp if the time is valid; null if time is invalid
*/
public WindowWrap<T> currentWindow(long timeMillis) {
if (timeMillis < 0) {
return null;
}
int idx = calculateTimeIdx(timeMillis);
// Calculate current bucket start time.
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 (array.compareAndSet(idx, null, window)) {
// 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));
}
}
}
我们再看看
FlowSlot的实现
获取时间窗口跟阈值进行判断 checker.checkFlow(ruleProvider, resource, context, node, count, prioritized);
有三个对应的实现，有金币算法，漏斗算法，限流算法 rule.getRater().canPass(selectedNode, acquireCount, prioritized)
再看看
DegradeSlot的实现
也是各种阈值判断，不通过就抛出异常。 DegradeRuleManager.checkDegrade(resourceWrapper, context, node, count);