一，sentinel源码分析
首先查看这个引入的依赖，
<!-- sentinel依赖--> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId> <version>2021.1</version> </dependency>
接下来看这个外部库里面对应的jar包，然后主要看里面的这个spring.factories这个文件
这个factories的文件里面的内容如下，因此主要分析这个带有autoConfiguration的这个类。
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ com.alibaba.cloud.sentinel.SentinelWebAutoConfiguration,\ com.alibaba.cloud.sentinel.SentinelWebFluxAutoConfiguration,\ com.alibaba.cloud.sentinel.endpoint.SentinelEndpointAutoConfiguration,\ com.alibaba.cloud.sentinel.custom.SentinelAutoConfiguration,\ com.alibaba.cloud.sentinel.feign.SentinelFeignAutoConfiguration org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=\ com.alibaba.cloud.sentinel.custom.SentinelCircuitBreakerConfiguration
二，SentinelAutoConfiguration
1，aop的底层实现
aop依赖与ioc，在生产bean并进行实例化之前，先通过bean的第一个后置处理器找到所有在类上面加@AspectJ这个注解的所有类，并在这个类的里面找到所有的befeore，after等注解的方法，每一个before，after等都会生成一个对应的advisor，每个advisor包括advise和pointcut，advise主要是用来作为一个增强器的使用，pointcut是为了进行匹配，匹配成功才进行最终的动态代理的生成。最后获取到所有的advisors，由于可能有大量的advisor，因此在bean的最后一个后置处理器才对这些所有的advisor进行处理，即在bean进行初始化之后才进行处理。最后会去循环遍历这些advisors，通过advisors里面封装的pointcut和生成的advisor进行比较，如果匹配成功，则说明bean需要创建动态代理。主要是通过责任链的方式实现
2，SentinelAutoConfiguration源码分析
其主类如下，有一个@ConditionalOnProperty这个注解，这个主要是在springboot在整个容器启动的时候会进行一个自动加载。
@Configuration(proxyBeanMethods = false) @ConditionalOnProperty(name = "spring.cloud.sentinel.enabled", matchIfMissing = true) @EnableConfigurationProperties(SentinelProperties.class) public class SentinelAutoConfiguration { }
在这个类里面，有一个@Bean的一个注解，主要是为了向这个容器里面注入一个SentinelResourceAspect的这个实例。
@Bean @ConditionalOnMissingBean public SentinelResourceAspect sentinelResourceAspect() { return new SentinelResourceAspect(); }
SentinelResourceAspect这个类如下，在这个方法上有一个**@AspectJ**的这个注解，因此可以知道这个类使用了这个aop来实现，
@Aspect public class SentinelResourceAspect extends AbstractSentinelAspectSupport{ }
在这个类里面，有两个方法，上面有一些@Around，@Pointcut的注解，主要是为了生成一些advisor
@Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)") public void sentinelResourceAnnotationPointcut() { } @Around("sentinelResourceAnnotationPointcut()") public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable{ try { //sentinel的核心代码 entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs()); Object result = pjp.proceed(); return result; } }
通过这个entry方法最终可以进入到一个entryWithPriority方法里面
private Entry entryWithPriority(){ //责任链模式 ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper); //通过这个链路调用链路上的某一个槽点 chain.entry(context, resourceWrapper, null, count, prioritized, args); }
接下来主要查看这个lookProcessChain的这个方法，里面主要是初始化了一个链条
ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper){ ProcessorSlotChain chain = chainMap.get(resourceWrapper); if (chain == null){ chain = SlotChainProvider.newSlotChain(); } return chain; }
接下来主要看这个newSlotChain方法，主要是看这个实例化链路的具体过程。
public static ProcessorSlotChain newSlotChain(){ slotChainBuilder = SpiLoader.loadFirstInstanceOrDefault(SlotChainBuilder.class, DefaultSlotChainBuilder.class); return slotChainBuilder.build(); }
然后进入这个build方法，看里面的具体build的构建流程
public ProcessorSlotChain build(){ ProcessorSlotChain chain = new DefaultProcessorSlotChain(); //通过这个spi机制进行一个解耦 //将一些类进行一个加载 List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class); for (ProcessorSlot slot : sortedSlotList) { if (!(slot instanceof AbstractLinkedProcessorSlot)) { continue; } chain.addLast((AbstractLinkedProcessorSlot<?>) slot); } return chain; }
这个链路的添加方法如下，主要是将这些slot槽点加入到这个链条里面，主要是通过这个责任链的模式实现。
//头插法 @Override public void addFirst(AbstractLinkedProcessorSlot<?> protocolProcessor) { protocolProcessor.setNext(first.getNext()); first.setNext(protocolProcessor); if (end == first) { end = protocolProcessor; } } //尾插法 @Override public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor) { end.setNext(protocolProcessor); end = protocolProcessor; }
再链路构建好了之后，就会去调用这个链路里面的slot槽点
接下来主要查看这个DefaultProcessorSlotChain默认类里面的槽点的entry方法
@Override public void entry(){ throws Throwable { first.transformEntry(context, resourceWrapper, t, count, prioritized, args); }
然后在这个transformEntry方法里面，又有一个entry方法，可以发现来里面可以调用多个槽点
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); }
然后可以进入这个NodeSelectorSlot这个类里面，主要是构建一个资源的请求路径，然后通过这个fireEntry的方法，来实现各个slot的结点的遍历
public void entry 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); } } } context.setCurNode(node); fireEntry(context, resourceWrapper, node, count, prioritized, args); }
3，FlowSlot流控规则
通过上面的源码分析可知，这个sentinel主要是通过这个责任链的方式实现，这个链路上面主要会有大量的slot槽点，通过会通过这个fireEntry方法按顺序来依次遍历一下的槽点，接下来主要分析一下这个FlowSlot的这个限流的这个槽点，每一个槽点就是对应的一个规则
在进入这个规则之后，首先会有一个流量的校验规则，通过这个checkFlorw的方法实现。
@Override 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); }
可以查看这个checkFlow的这个规则，会获取所有的这个流控规则，如一些阈值的设置，流控模式，流控效果等，如果校验不通过，那么就会直接抛异常
public void checkFlow(){ Collection<FlowRule> rules = ruleProvider.apply(resource.getName()); if (rules != null) { for (FlowRule rule : rules) { if (!canPassCheck(rule, context, node, count, prioritized)) { throw new FlowException(rule.getLimitApp(), rule); } } } }
那么主要是通过这个canPassCheck方法来判断校验是否通过
public boolean canPassCheck(){ //返回一个本地的检测 return passLocalCheck(rule, context, node, acquireCount, prioritized) }
在这个passLocalCheck检测方法里面，会有一个具体的一个规则的判断，会通过这个用户在界面上的选项来进行一个判断，如流控模式是选的直连，关联还是链路
private static boolean passLocalCheck() { Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node); if (selectedNode == null) { return true; } return rule.getRater().canPass(selectedNode, acquireCount, prioritized); }
然后在获取了所有的用户的值之后，会进行一个最终的判断，如判断这个qps或者并发数是否超过这个阈值，其逻辑主要是在这个canPass这个方法里面实现。如果不能通过，那么会直接返回一个false，并且直接会抛一个异常，链路直接断掉。并且这个异常可以进入一个HandlerBlockExecption。
public boolean canPass(Node node, int acquireCount, boolean prioritized) { int curCount = avgUsedTokens(node); if (curCount + acquireCount > count) { if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) { long currentTime; long waitInMs; currentTime = TimeUtil.currentTimeMillis(); waitInMs = node.tryOccupyNext(currentTime, acquireCount, count); if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) { node.addWaitingRequest(currentTime + waitInMs, acquireCount); node.addOccupiedPass(acquireCount); sleep(waitInMs); throw new PriorityWaitException(waitInMs); } } return false; } return true; }
如果没有出现异常，那么就会继续执行这个以下的业务流程。
4，DegradeSlot熔断规则
类似于这种电路的熔断器一样，主要有三种状态，分别是打开状态，半开状态和关闭状态。
@SpiOrder(-1000) public class DegradeSlot extends AbstractLinkedProcessorSlot<DefaultNode> { }
在这个方法里面，有一个entry方法，就是加入这个槽点的一个方法
@Override public void entry() throws Throwable { performChecking(context, resourceWrapper); fireEntry(context, resourceWrapper, node, count, prioritized, args); }
然后这个performChecking方法就是这个熔断器的核心方法，会去获取所有的已经配置的熔断规则，然后去校验每一个断路规则。
void performChecking(Context context, ResourceWrapper r) throws BlockException { //获取全部的配置的熔断规则 List<CircuitBreaker> cir = DegradeRuleManager.getCircuitBreakers(r.getName()); if (cir == null || cir.isEmpty()) { return; } for (CircuitBreaker cb : cir) { if (!cb.tryPass(context)) { throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule()); } } }
然后主要查看这个tryPass的这个方法，如果没有触发这个断路器，那么直接返回true
@Override public boolean tryPass(Context context) { // Template implementation. if (currentState.get() == State.CLOSED) { return true; } if (currentState.get() == State.OPEN) { // For half-open state we allow a request for probing. return retryTimeoutArrived() && fromOpenToHalfOpen(context); } return false; }
如果出现这个熔断，那么就会进入这个retryTimeoutArrived方法，如果当前时间大于这个熔断的时间，就会进入一个半开状态
protected boolean retryTimeoutArrived() { return TimeUtil.currentTimeMillis() >= nextRetryTimestamp; } protected void updateNextRetryTimestamp() { this.nextRetryTimestamp = TimeUtil.currentTimeMillis() + recoveryTimeoutMs; }
半开状态，主要是通过这个fromOpenToHalfOpen实现，会给这个状态做一个标记，作为一个半开状态。就是允许这个客户端发起一次这个请求，看执行结果是否通过，如果通过，则恢复成关闭状态，否则回到之前的全开状态。
protected boolean fromOpenToHalfOpen(Context context) { if (currentState.compareAndSet(State.OPEN, State.HALF_OPEN)) { notifyObservers(State.OPEN, State.HALF_OPEN, null); Entry entry = context.getCurEntry(); entry.whenTerminate(new BiConsumer<Context, Entry>() { @Override public void accept(Context context, Entry entry) { if (entry.getBlockError() != null) { currentState.compareAndSet(State.HALF_OPEN, State.OPEN); notifyObservers(State.HALF_OPEN, State.OPEN, 1.0d); } } }); return true; } return false; }
在关闭时有一个这个exit的这个方法，这里面实现了这个状态的切换
public void exit(){ if (curEntry.getBlockError() == null) { for (CircuitBreaker circuitBreaker : circuitBreakers) { circuitBreaker.onRequestComplete(context); } } }
然后在这个onRequestComplete方里面有一个核心方法handleStateChangeWhenThresholdExceeded
public void onRequestComplete(Context context){ handleStateChangeWhenThresholdExceeded(rt); }
然后进入这个handleStateChangeWhenThresholdExceeded的这个方法，里面有具体的状态切换。就是在进入这个半开状态之后，会再发起一次请求，如果请求时间超过这个阈值，那么又会重新进入这个全开状态，如果没有超过这个阈值，那么就会进入这个关闭状态。
private void handleStateChangeWhenThresholdExceeded(long rt) { if (currentState.get() == State.OPEN) { return; } if (currentState.get() == State.HALF_OPEN) { if (rt > maxAllowedRt) { fromHalfOpenToOpen(1.0d); } else { fromHalfOpenToClose(); } return; } }
这个半开到全开状态的转换，主要是通过这个cas比较与交换的这个算法实现，并且会重新更新这个滑动窗口的这个时间
protected boolean fromHalfOpenToOpen(double snapshotValue) { if (currentState.compareAndSet(State.HALF_OPEN, State.OPEN)) { updateNextRetryTimestamp(); notifyObservers(State.HALF_OPEN, State.OPEN, snapshotValue); return true; } return false; }
三，总结
首先会在这个服务端里面配置一些规则，然后会将这些规则推送给客户端，客户端将这些规则存在一个内存的一个链条里面，在调用所有的处理器的时候，都会经过这个链条。在经过这个链条的过程中，如果有异常抛出，会根据异常的种类，调用对应的降级方法。
