国产欧美在线观看,一区二区免费视频,人人九九精

一篇帶給你Sentinel 流控原理

我們在項目中添加 Spring Cloud Sentinel 依賴添加后 spring-cloud-starter-alibaba-sentinel 在 Spring-Boot 啟動的過程中回去初始化 spring.factories 中的配置信息，如：SentinelWebAutoConfiguration 、SentinelAutoConfiguration 等配置文件來初始化

再講代碼之前我先聲明一下我的版本號sentinel 1.8.0 。后續的所有內容均基于該版本進行

@ResoureSetinel 工作原理

配置流控規則我們最簡單的方式就是通過 @ResoureSetinel 的方式來管理，該注解可以直接定義流控規則、降級規則。下面是一個簡單的使用例子：

@SentinelResource(value = "ResOrderGet",
fallback = "fallback",
fallbackClass = SentinelResourceExceptionHandler.class,
blockHandler = "blockHandler",
blockHandlerClass = SentinelResourceExceptionHandler.class
)
@GetMapping("/order/get/{id}")
public CommonResult<StockModel> getStockDetails(@PathVariable Integer id) {
StockModel stockModel = new StockModel();
stockModel.setCode("STOCK==>1000");
stockModel.setId(id);
return CommonResult.success(stockModel);
}

如果大家熟悉 Spring 相關的組件大家都可以想到，這里多半是通過Spring Aop. 的方式來攔截 getStockDetails 方法。我們先看看SentinelAutoConfiguration 配置文件，我們可以找到 SentinelResourceAspect Bean 的定義方法。

@Bean
@ConditionalOnMissingBean
public SentinelResourceAspect sentinelResourceAspect() {
return new SentinelResourceAspect();
}

讓后我們再來看看 SentinelResourceAspect 具體是怎么處理的，源碼如下：

// 定義 Pointcut
@Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")
public void sentinelResourceAnnotationPointcut() {
}
// Around 來對被標記 @SentinelResource 注解的方法進行處理
@Around("sentinelResourceAnnotationPointcut()")
public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {
Method originMethod = resolveMethod(pjp);
// 獲取注解信息
SentinelResource annotation = originMethod.getAnnotation(SentinelResource.class);
// 獲取資源名稱
String resourceName = getResourceName(annotation.value(), originMethod);
EntryType entryType = annotation.entryType();
int resourceType = annotation.resourceType();
Entry entry = null;
try {
// 執行 entry
entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());
// 執行業務方法
Object result = pjp.proceed();
// 返回
return result;
} catch (BlockException ex) {
// 處理 BlockException
return handleBlockException(pjp, annotation, ex);
} catch (Throwable ex) {
Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();
// The ignore list will be checked first.
if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {
throw ex;
}
if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {
traceException(ex);
// 處理降級
return handleFallback(pjp, annotation, ex);
}
// No fallback function can handle the exception, so throw it out.
throw ex;
}
}

我們總結一下， @SentinelResource 的執行過程，首先是通過 Aop 進行攔截，然后通過 SphU.entry 執行對應的流控規則，最后調用業務方法。如果觸發流控規則首先處理流控異常 BlockException 然后在判斷是否有服務降級的處理，如果有就調用 fallback 方法。通過 handleBlockException 、handleFallback 進行處理。

責任鏈模式處理流控

通過上面的梳理，我們知道對于流控的過程，核心處理方法就是 SphU.entry 。在這個方法中其實主要就是初始化流控 Solt 和執行 Solt. 在這個過程中會對：簇點定義、流量控制、熔斷降級、系統白名單等頁面功能進行處理。

1. 初始化責任鏈

下面是初始化 Solt 的核心代碼在 SphU.entryWithPriority

// 刪減部分代碼
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
throws BlockException {
// 初始化責任鏈
ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
Entry e = new CtEntry(resourceWrapper, chain, context);
try {
// 執行 entry
chain.entry(context, resourceWrapper, null, count, prioritized, args);
} catch (BlockException e1) {
e.exit(count, args);
// 異常拋出，讓 SentinelResourceAspect.invokeResourceWithSentinel 統一處理
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;
}

通過 lookProcessChain 方法我逐步的查找，我們可以看到最終的責任鏈初始化類，默認是 DefaultSlotChainBuilder

public class DefaultSlotChainBuilder implements SlotChainBuilder {
@Override
public ProcessorSlotChain build() {
ProcessorSlotChain chain = new DefaultProcessorSlotChain();
// Note: the instances of ProcessorSlot should be different, since they are not stateless.
// 通過 SPI 去加載所有的 ProcessorSlot 實現，通過 Order 排序
List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);
for (ProcessorSlot slot : sortedSlotList) {
if (!(slot instanceof AbstractLinkedProcessorSlot)) {
RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() + ") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
continue;
}
// 添加到 chain 尾部
chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
}
return chain;
}
}

2. 責任鏈的處理過程

我們可以通過斷點的方式來查看在 sortedSlotList 集合中所有的 solt 順序如下圖所示:

一篇帶給你Sentinel 流控原理

我們可以通過如下的順序進行逐個的簡單的分析一下

NodeSelectorSolt
CusterBuilderSolt
LogSlot
StatisicSlot
AuthoritySolt
SystemSolts
ParamFlowSolt
FlowSolt
DegradeSlot

對于 Sentinel 的 Slot 流控協作流程可以參考官方給出的文檔, 如下圖所示：

一篇帶給你Sentinel 流控原理

FlowSolt 流控

通過 NodeSelectorSolt、CusterBuilderSolt、StatisicSlot 等一系列的請求數據處理，在 FlowSolt會進入流控規則，所有的 Solt 都會執行 entry 方法, 如下所示

// FlowSolt 的 entry 方法
@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);
}

在后續的流程中，會執進行判斷具體的流控策略，默認是快速失敗，會執行 DefaultController 方法。

// DefaultController
@Override
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
// 當前資源的調用次數
int curCount = avgUsedTokens(node);
// 當前資源的調用次數 + 1 > 當前閾值
if (curCount + acquireCount > count) {
// 刪減比分代碼
// 不通過
return false;
}
// 通過
return true;
}
private int avgUsedTokens(Node node) {
if (node == null) {
return DEFAULT_AVG_USED_TOKENS;
}
return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
}

如果上面返回不通過會回到，那么會拋出 FlowException

public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,
Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
if (ruleProvider == null || resource == null) {
return;
}
Collection<FlowRule> rules = ruleProvider.apply(resource.getName());
if (rules != null) {
for (FlowRule rule : rules) {
if (!canPassCheck(rule, context, node, count, prioritized)) {
// 流控規則不通過，會拋出 FlowException
throw new FlowException(rule.getLimitApp(), rule);
}
}
}
}

然后會在 StatisticSlot 中增加統計信息，最后會拋出給 SentinelResourceAspect 進行處理，完成流控功能。我們再來看看這個異常信息，如果是BlockException 異常，會進入 handleBlockException 方法處理，如果是其他的業務異常首先會判斷是否有配置 fallback 處理如果有，就調用 handleFallback 沒有就繼續往外拋，至此完成流控功能

try {
entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());
Object result = pjp.proceed();
return result;
} catch (BlockException ex) {
return handleBlockException(pjp, annotation, ex);
} catch (Throwable ex) {
Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();
// The ignore list will be checked first.
if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {
throw ex;
}
if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {
traceException(ex);
return handleFallback(pjp, annotation, ex);
}
// No fallback function can handle the exception, so throw it out.
throw ex;
}

DegradeSlot 降級

斷路器的作用是當某些資源一直出現故障時，將觸發斷路器。斷路器不會繼續訪問已經發生故障的資源，而是攔截請求并返回故障信號。

Sentinel 在 DegradeSlot 這個 Slot 中實現了熔斷降級的功能，它有三個狀態 OPEN 、HALF_OPEN、CLOSED 以ResponseTimeCircuitBreaker RT 響應時間維度來分析，斷路器工作的過程。下面是一個標準斷路器的工作流程：

一篇帶給你Sentinel 流控原理

在 Sentinel 實現的源碼過程如下圖所示：

一篇帶給你Sentinel 流控原理

Sentinel 通過 Web 攔截器

Sentinel 在默認情況下，不使用 @ResourceSentinel 注解實現流控的時候， Sentinel 通過攔截器進行流控實現的。初始化類在 SentinelWebAutoConfiguration 它實現了 WebMvcConfigurer 接口，在 sentinelWebInterceptor 方法初始化 SentinelWebInterceptor 等 Bean。

@Bean
@ConditionalOnProperty(name = "spring.cloud.sentinel.filter.enabled",
matchIfMissing = true)
public SentinelWebInterceptor sentinelWebInterceptor(
SentinelWebMvcConfig sentinelWebMvcConfig) {
return new SentinelWebInterceptor(sentinelWebMvcConfig);
}

我們在 SentinelWebInterceptor 的核心方法 preHandle 中處理，這里面我們又可以看到 SphU.entry 熟悉的過程調用流控的責任鏈。由于邏輯都類似，此處不再多說。代碼如下：

public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
throws Exception {
try {
String resourceName = getResourceName(request);
if (StringUtil.isEmpty(resourceName)) {
return true;
}
if (increaseReferece(request, this.baseWebMvcConfig.getRequestRefName(), 1) != 1) {
return true;
}
// Parse the request origin using registered origin parser.
String origin = parseOrigin(request);
String contextName = getContextName(request);
ContextUtil.enter(contextName, origin);
Entry entry = SphU.entry(resourceName, ResourceTypeConstants.COMMON_WEB, EntryType.IN);
request.setAttribute(baseWebMvcConfig.getRequestAttributeName(), entry);
return true;
} catch (BlockException e) {
try {
handleBlockException(request, response, e);
} finally {
ContextUtil.exit();
}
return false;
}
}