《一起学sentinel》六、Slot的子类及实现之FlowSlot和DegradeSlot-阿里云开发者社区

一、概述

在 Sentinel 里面，所有的资源都对应一个资源名称（resourceName），每次资源调用都会创建一个 Entry 对象。Entry 可以通过对主流框架的适配自动创建，也可以通过注解的方式或调用 SphU API 显式创建。Entry 创建的时候，同时也会创建一系列功能插槽（slot chain），这些插槽有不同的职责，例如:

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

下面是关系结构图

ProcessorSlot子类及实现类.png

二、FlowSlot分析

1.FlowSlot介绍

官方文档是这样描述FlowSlot的：

流量控制（flow control），其原理是监控应用流量的 QPS 或并发线程数等指标，当达到指定的阈值时对流量进行控制，以避免被瞬时的流量高峰冲垮，从而保障应用的高可用性。

FlowSlot 会根据预设的规则，结合前面 NodeSelectorSlot、ClusterBuilderSlot、StatisticSlot 统计出来的实时信息进行流量控制。

限流的直接表现是在执行 Entry nodeA = SphU.entry(resourceName) 的时候抛出 FlowException 异常。FlowException 是 BlockException 的子类，您可以捕捉 BlockException 来自定义被限流之后的处理逻辑。

同一个资源可以创建多条限流规则。FlowSlot 会对该资源的所有限流规则依次遍历，直到有规则触发限流或者所有规则遍历完毕。

一条限流规则主要由下面几个因素组成，我们可以组合这些元素来实现不同的限流效果：

resource：资源名，即限流规则的作用对象
count: 限流阈值
grade: 限流阈值类型（QPS 或并发线程数）
limitApp: 流控针对的调用来源，若为 default 则不区分调用来源
strategy: 调用关系限流策略

controlBehavior: 流量控制效果（直接拒绝、Warm Up、匀速排队）

这个是默认行为，超出的请求会被拒绝。并抛出FlowException。


- ```java
服务升温Warmup ({@code RuleConstant.CONTROL_BEHAVIOR_WARM_UP})
如果系统的负载已经低了一段时间，和大量的请求到来时，系统可能无法处理所有这些请求。
但是，如果我们稳定地增加传入请求，系统就会升温，最终能够处理所有的请求。
此预热期可通过在流规则中设置字段{@code warmupperiods}来配置。

这个策略严格控制请求之间的间隔。
换句话说，它允许请求以稳定、统一的速率通过。
该策略是leaky bucket的实现。
它用于以稳定的速率处理请求，经常用于突发流量(例如消息处理)。


我们可以根据以下命令获取到样例图

//命令：

curl http://localhost:8719/tree
//样例图：
idx id thread pass blocked success total aRt 1m-pass 1m-block 1m-all e
2 abc647 0 460 46 46 1 27 630 276 897 0


其中：

- `thread`： 代表当前处理该资源的**并发**数；
- `pass`： 代表一**秒**内到来到的**请求**；
- `blocked`： 代表一**秒**内被流量**控制**的请求数量；
- `success`： 代表一**秒**内**成功**处理完的请求；
- `total`： 代表到一**秒**内到来的请求以及被**阻止**的请求**总和**；
- RT： 代表一秒内该资源的**平均响应时间**；
- `1m-pass`： 则是一**分钟**内到来的**请求**；
- `1m-block`： 则是一**分钟**内被**阻止**的请求；
- `1m-all`： 则是一**分钟**内到来的请求和**被阻止**的请求的**总和**；
- `(e)exception`： 则是一秒内业务本身**异常**的**总和**。

#### 2.源码解读

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

}

@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {

fireExit(context, resourceWrapper, count, args);

}


1.在`entry`阶段，执行了一个校验方法.

void checkFlow(ResourceWrapper resource, Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {

checker.checkFlow(ruleProvider, resource, context, node, count, prioritized);

}


2.我们可以看到这里有一个多出来的关键参数“`ruleProvider`”，我们看看这个多出来参数的实现。

private final Function> ruleProvider = new Function>() {

@Override
public Collection<FlowRule> apply(String resource) {
    // Flow rule map should not be null.
    Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();
    return flowRules.get(resource);
}

};


3.首先会在全局的`FlowRuleManager`中获取全局的`FlowRuleMap`，然后根据我们的唯一判断准则“`resource`”获取对应的`FlowRuleList`。

4.接下来我们看看`checkFlow`方法。

public void checkFlow(Function> 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)) {
            throw new FlowException(rule.getLimitApp(), rule);
        }
    }
}

}


5.首先对`resource`和`ruleList`进行了判断，如果为空着直接跳过校验。接着取出`ruleList`，分别判断每一个判断是否满足，如果不满足， `throw new FlowException(rule.getLimitApp(), rule)`;

public boolean canPassCheck(/@NonNull/ FlowRule rule, Context context, DefaultNode node, int acquireCount,boolean prioritized) {

String limitApp = rule.getLimitApp();
if (limitApp == null) {
return true;
}

if (rule.isClusterMode()) {
return passClusterCheck(rule, context, node, acquireCount, prioritized);
}

return passLocalCheck(rule, context, node, acquireCount, prioritized);
}

}


6.我们进入内层方法可以发现，这里区分了集群模式和本地模式，就算选择了集群模式后续代码中也会重新进行集群模式的校验，如果校验失败则会降级退回到本地模式。

static Node selectNodeByRequesterAndStrategy(/@NonNull/ FlowRule rule, Context context, DefaultNode node) {

// The limit app should not be empty.
String limitApp = rule.getLimitApp();
int strategy = rule.getStrategy();
String origin = context.getOrigin();

if (limitApp.equals(origin) && filterOrigin(origin)) {
    if (strategy == RuleConstant.STRATEGY_DIRECT) {
        // Matches limit origin, return origin statistic node.
        return context.getOriginNode();
    }

    return selectReferenceNode(rule, context, node);
} else if (RuleConstant.LIMIT_APP_DEFAULT.equals(limitApp)) {
    if (strategy == RuleConstant.STRATEGY_DIRECT) {
        // Return the cluster node.
        return node.getClusterNode();
    }

    return selectReferenceNode(rule, context, node);
} else if (RuleConstant.LIMIT_APP_OTHER.equals(limitApp)
           && FlowRuleManager.isOtherOrigin(origin, rule.getResource())) {
    if (strategy == RuleConstant.STRATEGY_DIRECT) {
        return context.getOriginNode();
    }

    return selectReferenceNode(rule, context, node);
}

return null;

}


7.现在我们终于进入到了`strategy`的逻辑，这里主要逻辑是判断在不同的**limitApp**下，如指定类型，集群，其他以及**STRATEGY_DIRECT**流程，如果全部匹配失败后会进入到`selectReferenceNode`，这里包含了**STRATEGY_RELATE**流程以及**STRATEGY_CHAIN**流程。

接下来则到了最后一块：controlBehavior

接下来就是根据数据与流量控制规则进行判断，是否通过。

![image.png](//p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/6119fd35ce3a4d3c9538a16bc7da6ff2~tplv-k3u1fbpfcp-zoom-1.image)

8.这个控制器有四个实现类，对应了`flow`的最后一个关键因子“**controlBehavior**”

* `DefaultController`默认的节流控制器(立即拒绝策略`Immediately`)。
* `RateLimiterController`稳定匀速的令牌桶方式（匀速排队`Uniform` `Rate`）
* `WarmUpController`(服务升温`Warmup`)
* `WarmUpRateLimiterController`(服务升温+令牌桶`Warmup+RateLimiter`)

作为`sentinel` 的核心限流控制器，就和我们使用的方式一样，预先设置了大量的对应具体资源的规则，规则会在初始化时被注册为一个`map`，这里我们可以看到`sentinel`使用了`CopyOnWrite`的思想去操作`flow`的`map`。进行`grade`、`strategy`、`controlBehavior`的多维度组合限流后，完整的实现了限流的功能。





## 三、DegradeSlot分析

#### 1.DegradeSlot介绍

官方文档是这样描述`DegradeSlot`的：致力于断路器。个人认为这个是`sentinel`比起一般的网关，最具差异的地方，既有丰富的限流，又提供了熔断的能力。

#### 2.源码解读

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {

performChecking(context, resourceWrapper);

fireEntry(context, resourceWrapper, node, count, prioritized, args);

}

@Override
public void exit(Context context, ResourceWrapper r, int count, Object... args) {

Entry curEntry = context.getCurEntry();
if (curEntry.getBlockError() != null) {
    fireExit(context, r, count, args);
    return;
}
List<CircuitBreaker> circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
if (circuitBreakers == null || circuitBreakers.isEmpty()) {
    fireExit(context, r, count, args);
    return;
}

if (curEntry.getBlockError() == null) {
    // passed request
    for (CircuitBreaker circuitBreaker : circuitBreakers) {
        circuitBreaker.onRequestComplete(context);
    }
}

fireExit(context, r, count, args);

}


1.我们可以看到，这里的entry里面调用了私有的performChecking方法，和flow一样，我们先看看CircuitBreaker的对象结构。

public interface CircuitBreaker {

/**
   *获取相关的断路规则。
 */
DegradeRule getRule();

/**
 * 仅当调用时该调用可用时，才获取该调用的权限。
 *
 * @param context 当前调用的上下文
 * @return true 如果获得了权限，则使用return false
 */
boolean tryPass(Context context);

/**
 * 获取断路器的通过状态。
 */
State currentState();

/**
 * 用上下文记录一个已完成的请求，并对断路器进行状态转换
 */
void onRequestComplete(Context context);

/**
 * Circuit breaker state.
 */
enum State {
    /**
     * 在{@code OPEN}状态下，所有请求都将被拒绝，直到下一个恢复时间点。
     */
    OPEN,
    /**
     *在{@code HALF_OPEN}状态下，断路器允许“探测”调用。
      *如果调用异常，根据策略(例如，它是缓慢的)，断路器
     *将重新转换为{@code OPEN}状态，等待下一个恢复时间点;
     *否则，该资源将被视为“恢复”和断路器
     *将停止切断请求并转换为{@code CLOSED}状态。
     */
    HALF_OPEN,
    /**
     *在{@code CLOSED}状态中，允许所有请求。当当前度量值超过阈值时，
     *断路器将转换为{@code OPEN}状态。
     */
    CLOSED
}

}


整段都在维护这个CircuitBreaker的state，我们可以看到这里不再是resource，而是context，这里的状态决定了请求是否能够通过，如果在这个最低成的slot就被拦截并拒绝，那么可以理解为是不需要再次限流的（虽然会统计数量）

void performChecking(Context context, ResourceWrapper r) throws BlockException {

List<CircuitBreaker> circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
if (circuitBreakers == null || circuitBreakers.isEmpty()) {
    return;
}
for (CircuitBreaker cb : circuitBreakers) {
    if (!cb.tryPass(context)) {
        throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule());
    }
}

}


2.performChecking主要的逻辑就是从当前DegradeRuleManager中获取resource对应的熔断规则，如果需要进行熔断则`throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule())`;





## 四、小结

本期我们讲述了Slot的子类`FlowSlot`和`DegradeSlot`的基本实现原理。

现在建立我们的知识树



#### 实例化DefaultNode和ClusterNode，创建结构树

------

创建上下文时，首先会在`NodeSelectorSlot`中判断是否有`DefaultNode`。

如果没有则新增一个基于`resource`的`DefaultNode`，然后执行下一个`slot`。

下一个`slot`是`ClusterBuilderSlot`，`ClusterBuilderSlot`会判断是否有对应的`ClusterNode`，如果没有则新增一个基于resource的`ClusterNode`并继续下一个流程（`slot`）。

总结来说，这个两个`slot`奠定了一个基于`resource`进行全局控制的基调。



#### 进行信息收集

------

`LogSlot`在`DefaultNode`和`ClusterNode`初始化后，作为业务实例模块的分界点，收集全局异常并处理。

`StatisticSlot`作为全局统计的实例，依托于`ClusterNode`，将全局的`RT`, `QPS`, `thread` `count` 等等信息存放在`clusterNodeMap`里面。



#### 进行权限校验及系统级限流

---

在树结构和信息收集的slot建立完毕后，开始业务逻辑的实现，首先实现的就是AuthoritySlot的黑白名单能力，依托sentinel的resource的定义，我们很简单就可以拿到关于resource的authorityRules，将对应的rules取出后，以此进行黑、白名单判断，也可以理解为一种权限级别的限流措施。

SystemSlot则是全统计的全局限流，从调用点origins级别的配置中读取了配置好的限流措施，在下一个slot实现前完成了所有的判断，如qps，线程数，成功访问数，RT,CPU状态。如果出现异常，则throws BlockException，交给之前的slot去处理相应逻辑。到这里，一个基础的限流框架已经基本实现。



#### 进行限流和熔断

---

当所有的配置项已经配置完毕，权限级别和系统级别的限流做完，现在轮到了最后的两个slot。

flowslot和DegradeSlot分别对应了我们配置的限流flow和配置的熔断机制。

到这里，一个成熟的分布式网关已经完成，我们的sentinel的完整功能已经叙述完毕。

《一起学sentinel》六、Slot的子类及实现之FlowSlot和DegradeSlot

一、概述

二、FlowSlot分析

1.FlowSlot介绍

热门文章

最新文章

相关课程

相关电子书

相关实验场景