Sentinel是什么？
官网地址：https://github.com/alibaba/Sentinel/wiki/介绍
Sentinel 是面向分布式服务架构的流量控制组件，主要以流量为切入点，从限流、流量整形、熔断降级、系统负载保护、热点防护等多个维度来帮助开发者保障微服务的稳定性。
基本概念？
1.资源
2.规则
功能和设计理念？
1.流量控制
流量控制设计理念
并发数控制用于保护业务线程池不被慢调用耗尽。例如，当应用所依赖的下游应用由于某种原因导致服务不稳定、响应延迟增加，对于调用者来说，意味着吞吐量下降和更多的线程数占用，极端情况下甚至导致线程池耗尽。Sentinel 并发控制不负责创建和管理线程池，而是简单统计当前请求上下文的线程数目（正在执行的调用数目），如果超出阈值，新的请求会被立即拒绝。
QPS流量控制
直接拒绝
Warm Up
匀速排队
这种方式主要用于处理间隔性突发的流量，例如消息队列。想象一下这样的场景，在某一秒有大量的请求到来，而接下来的几秒则处于空闲状态，我们希望系统能够在接下来的空闲期间逐渐处理这些请求，而不是在第一秒直接拒绝多余的请求。
2.熔断降级
熔断降级设计理念
熔断策略Sentinel 提供以下几种熔断策略：
2.异常比例 (ERROR_RATIO)：当单位统计时长（statIntervalMs）内请求数目大于设置的最小请求数目，并且异常的比例大于阈值，则接下来的熔断时长内请求会自动被熔断。经过熔断时长后熔断器会进入探测恢复状态（HALF-OPEN 状态），若接下来的一个请求成功完成（没有错误）则结束熔断，否则会再次被熔断。异常比率的阈值范围是 [0.0, 1.0]，代表 0% - 100%。
3.异常数 (ERROR_COUNT)：当单位统计时长内的异常数目超过阈值之后会自动进行熔断。经过熔断时长后熔断器会进入探测恢复状态（HALF-OPEN 状态），若接下来的一个请求成功完成（没有错误）则结束熔断，否则会再次被熔断。
源码分享
● sentinel-adapter 适配器模块，主要实现了对一些常见框架的适配
主要分享限流和熔断比较关键的实现：
限流入口
根据Sentinel提供的demo模块可以快速找到限流入口，就是SphU.entry(KEY)代码，此处的key唯一确定限流规则，Sentinel会根据这个key去获取对应的限流规则；客户端请求达到限流规则，抛出BlockException，进入catch模块；在finally内退出entry释放资源。
Entry entry = null;
try {
entry = SphU.entry(KEY);
//被保护的资源
.......
} catch (BlockException blockException) {
// 接口被限流的时候, 会进入到这里
System.out.println("---接口被限流了---, exception: ");
return ResponseEntity.ok("接口限流, 返回空");
} finally {
// 释放资源
if (entry != null) {
entry.exit();
}
}
entry实现：代 码进来先进行了必要参数的校验，此处的ResourceWrapper就是根据前面的key和EntryType构建的资源包装类
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args) throws BlockException {
// 从 ThreadLocal 中获取 Context 实例
Context context = ContextUtil.getContext();
// 如果是 NullContext，那么说明 context name 超过了 2000 个，参见 ContextUtil.trueEnter
// 这个时候，Sentinel 不再接受处理新的 context 配置，也就是不做这些新的接口的统计、限流熔断等
if (context instanceof NullContext) {
return new CtEntry(resourceWrapper, null, context);
}
// 如果我们不显式调用 ContextUtil.enter，这里会进入到默认的 context 中
if (context == null) {
context = MyContextUtil.myEnter(Constants.CONTEXT_DEFAULT_NAME, "", resourceWrapper.getType());
}
// Sentinel 的全局开关，Sentinel 提供了接口让用户可以在 dashboard 开启/关闭
if (!Constants.ON) {
return new CtEntry(resourceWrapper, null, context);
}
// 设计模式中的责任链模式。
// 下面这行代码用于构建一个责任链，入参是 resource，前面我们说过资源的唯一标识是 resource name
ProcessorSlot chain = lookProcessChain(resourceWrapper);
// 根据 lookProcessChain 方法，我们知道，当 resource 超过 Constants.MAX_SLOT_CHAIN_SIZE，
// 也就是 6000 的时候，Sentinel 开始不处理新的请求，这么做主要是为了 Sentinel 的性能考虑
if (chain == null) {
return new CtEntry(resourceWrapper, null, context);
}
// 执行这个责任链。如果抛出 BlockException，说明链上的某一环拒绝了该请求，
// 把这个异常往上层业务层抛，业务层处理 BlockException 应该进入到熔断降级逻辑中
Entry e = new CtEntry(resourceWrapper, chain, context);
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);
}
return e;
}
构建执行链条
根据资源包装类从链map中获取执行链，获取到直接返回；如果没有获取到说明是第一次构建该资源的执行链
ProcessorSlot lookProcessChain(ResourceWrapper resourceWrapper) {
ProcessorSlotChain chain = chainMap.get(resourceWrapper);
if (chain == null) {
synchronized (LOCK) {
chain = chainMap.get(resourceWrapper);
if (chain == null) {
// 根据 lookProcessChain 方法，我们知道，当 resource 超过 Constants.MAX_SLOT_CHAIN_SIZE，
// 也就是 6000 的时候，Sentinel 开始不处理新的请求，这么做主要是为了 Sentinel 的性能考虑
if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
return null;
}
chain = SlotChainProvider.newSlotChain();
Map newMap = new HashMap(chainMap.size() + 1);
newMap.putAll(chainMap);
newMap.put(resourceWrapper, chain);
chainMap = newMap;
}
}
}
return chain;
}
构建该资源对应的资源链，具体构建链代码如下：
@Override
public ProcessorSlotChain build() {
ProcessorSlotChain chain = new DefaultProcessorSlotChain();
List 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.addLast((AbstractlinkedProcessorSlot) slot);
}
return chain;
}
链条执行
关键类UML图如上，ProcessorSlot定义了链条上插槽方法，包括entry、fireEntry、exit、fireExit方法。链条上的所有插槽都通过抽象类AbstractlinkedProcessorSlot实现了ProcessoreSlot接口。
FlowSlot（限流插槽）实现如下：
@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);
}
限流插槽的真正的校验交给了FlowRuleChecker。先根据之前的key获取具体的限流规则，然后canPassCheck方法进行校验。代码如下：
public void checkFlow(Function> ruleProvider, ResourceWrapper resource,
Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
if (ruleProvider == null || resource == null) {
return;
}
Collection 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);
}
}
}
}
熔断降级
熔断操作同样是通过继承AbstractlinkedProcessorSlot实现ProcessorSlot接口的
AbstractlinkedProcessorSlot {
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)
throws Throwable {
DegradeRuleManager.checkDegrade(resourceWrapper, context, node, count);
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
fireExit(context, resourceWrapper, count, args);
}
}
熔断check：
public static void checkDegrade(ResourceWrapper resource, Context context, DefaultNode node, int count)
throws BlockException {
Set rules = degradeRules.get(resource.getName());
if (rules == null) {
return;
}
for (DegradeRule rule : rules) {
if (!rule.passCheck(context, node, count)) {
throw new DegradeException(rule.getLimitApp(), rule);
}
}
}
滑动窗口
sentinel的滑动窗口统计机制就是根据当前时间，获取对应的时间窗口，并更新该时间窗口中的各项统计指标（pass/block/rt等），这些指标被用来进行后续判断，比如限流、降级等；随着时间的推移，当前时间点对应的时间窗口是变化的，这时会涉及到时间窗口的初始化、复用等。可以说，sentinel上的功能所用到的数据几乎都是滑动窗口统计机制来维护和更新的。
既然要做控制，那么首先，Sentinel 就要先做统计，它要知道当前接口的 QPS 和并发是多少，进而判断一个新的请求能不能让它通过。
//保存最近的{@code INTERVAL}毫秒的统计信息
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);
//保存最近60秒的统计数据
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
从上面的代码可以知道，Sentinel 统计了 秒 和 分 两个维度的数据，下面我们简单说说实现类 ArrayMetric 的源码设计。
public class ArrayMetric implements Metric {
private final LeapArray data;
public ArrayMetric(int sampleCount, int intervalInMs) {
this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
}
public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy){ if (enableOccupy) {
this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs); } else {
this.data = new BucketLeapArray(sampleCount, intervalInMs);
}
} ...... }
ArrayMetric 的内部是一个 LeapArray，我们以分钟维度统计的使用来说，它使用子类 BucketLeapArray 实现。
public abstract class LeapArray {
protected int windowLengthInMs;
protected int sampleCount;
protected int intervalInMs;
protected final AtomicReferenceArray> array;
// 对于分钟维度的设置，sampleCount 为 60，intervalInMs 为 60 * 1000
public LeapArray(int sampleCount, int intervalInMs) {
// 单个窗口长度，这里是 1000ms
this.windowLengthInMs = intervalInMs / sampleCount;
// 一轮总时长 60,000 ms
this.intervalInMs = intervalInMs;
// 60 个窗口
this.sampleCount = sampleCount;
this.array = new AtomicReferenceArray<>(sampleCount);
}
// ......
}
它的内部核心是一个数组 array，它的长度为 60，也就是有 60 个窗口，每个窗口长度为 1 秒，刚好一分钟走完一轮。然后下一轮开启“覆盖”操作。
public WindowWrap currentWindow(long timeMillis) {
if (timeMillis < 0) { return null; }
// 获取窗口下标
int idx = calculateTimeIdx(timeMillis);
// 计算该窗口的理论开始时间
long windowStart = calculateWindowStart(timeMillis);
// 嵌套在一个循环中，因为有并发的情况
while (true) {
WindowWrap old = array.get(idx);
if (old == null) {
// 窗口未实例化的情况，使用一个 CAS 来设置该窗口实例
WindowWrap window = new WindowWrap(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
if (array.compareAndSet(idx, null, window)) {
return window;
} else {
// 存在竞争
Thread.yield();
}
} else if (windowStart == old.windowStart()) {
// 当前数组中的窗口没有过期
return old;
} else if (windowStart > old.windowStart()) {
// 该窗口已过期，重置窗口的值。使用一个锁来控制并发。
if (updateLock.tryLock()) {
try {
return resetWindowTo(old, windowStart);
} finally {
updateLock.unlock();
}
} else {
Thread.yield();
}
} else if (windowStart < old.windowStart()) {
// 正常情况都不会走到这个分支，异常情况其实就是时钟回拨，这里返回一个 WindowWrap 是容错
return new WindowWrap(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
}
}
}
获取数据，使用的是 values 方法，这个方法返回“有效的”窗口中的数据：
public List values(long timeMillis) {
if (timeMillis < 0) {
return new ArrayList();
}
int size = array.length();
List result = new ArrayList(size);
for (int i = 0; i < size; i++) {
WindowWrap windowWrap = array.get(i);
// 过滤掉过期数据
if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) { continue;
}
result.add(windowWrap.value());
}
return result;
}
// 判断当前窗口的数据是否是 60 秒内的
public boolean isWindowDeprecated(long time, WindowWrap windowWrap) {
return time - windowWrap.windowStart() > intervalInMs;
}
这个 values 方法很简单，就是过滤掉那些过期数据就可以了。
