spring cloud alibaba Sentinel：限流和服务熔断框架

spring cloud alibaba Sentinel是一个限流和服务熔断降级解决方案。Sentinel 是面向分布式、多语言异构化服务架构的流量治理组件，主要以流量为切入点，从流量路由、流量控制、流量整形、熔断降级、系统自适应过载保护、热点流量防护等多个维度来帮助开发者保障微服务的稳定性。

---

---

1、Sentinel限流

1.1、Sentinel限流的算法（限流源码中分别有讲解）：

• DefaultController：快速失败 ====》 滑动时间窗口
• RateLimiterController：排队等待 ====》漏斗算法
• WarmUpController：Warm Up ===》令牌桶算法

1.2、流控阈值类型

• QPS：允许1s内处理请求的数量，如果超过这个数，那么就说明请求数已经超过了系统在1s内能处理的最大数量，为了防止请求过多导致系统处理卡顿，需要限制请求数。
• 并发线程数：为每一个资源设置一个并发线程数，类似于信号量，每一个请求过来请求资源的时候，必须要获取到一个线程数才能继续访问资源，否则无法访问资源。

2、Sentinel服务熔断

1.2、Sentinel服务熔断方案（熔断源码中分别有讲解）：

• ResponseTimeCircuitBreaker：慢调用比例。
• ExceptionCircuitBreaker：异常比例

2.1、熔断策略：

• 异常数：1秒内发起10个请求，异常请求达到8个，就触发熔断。
• 平均响应时间：A服务调用B服务，平均响应时间超过10s，就触发熔断。
• 异常比例数：1秒内发起的请求，异常请求达到50%，就触发熔断。

2.2、熔断时间窗口：如果触发了熔断，在多久时间内，所有的请求过来都不能调用到真正的服务上边去，而是这段时间内，请求触发快速度失败策略。

2.3、实现方案：就是使用了一个大的try{}catch{},然后在catch中调用我们自行实现的熔断逻辑。

3、Sentinel简单的使用

spring-cloud和spring-cloud-alibaba项目，并需要整合nacos和gateway以及dubbo。

• pom.xml

<dependency>
    <groupId>com.alibaba.cloudgroupId>
    <artifactId>spring-cloud-alibaba-sentinel-gatewayartifactId>
dependency>
<dependency>
    <groupId>com.alibaba.cloudgroupId>
    <artifactId>spring-cloud-starter-alibaba-sentinelartifactId>
dependency>
<dependency>
    <groupId>com.alibaba.cspgroupId>
    <artifactId>sentinel-transport-simple-httpartifactId>
dependency>
<dependency>
    <groupId>com.alibaba.cspgroupId>
    <artifactId>sentinel-datasource-nacosartifactId>
dependency>

<dependency>
    <groupId>com.alibaba.cloudgroupId>
    <artifactId>spring-cloud-alibaba-sentinel-datasourceartifactId>
dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

• bootstrap.yml

spring:
  application:
    name: cloud-reading-gateway
  cloud:
    # sentinel配置
    sentinel:
      enabled: true
      eager: true
      transport:
        dashboard: 127.0.0.1:8080 # sentinel控制要地址
      datasource:
        flow-ds: # sentinel从nacos获取指定的流控规则 这个名字可以自定义，不重复就可以
          nacos:
            server-addr: 10.40.177.238:8848
            dataId: sentinel-floew
            groupId: DEFAULT_GROUP
            ruleType: flow # flow代表流程控制，degrade代表熔断规则
            username: nacos
            password: nacos
            data-type: json
        grade-ds: #sentinel从nacos获取指定的熔断规则 这个名字可以自定义，不重复就可以
          nacos:
            server-addr: 127.0.0.1:8848
            dataId: sentinel-degrade
            groupId: DEFAULT_GROUP
            ruleType: degrade # flow代表流程控制，degrade代表熔断规则
            username: nacos
            password: nacos
            data-type: json
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

• TestService

@Service
public class TestService {

    // 远程accoud服务
    @DubboReference
    ICloudAccoudService cloudAccoudService;

    // 使用SentinelResource注解，完成sentinel流控和熔断降级功能。
    // value：设置的是资源名称,可以自定义。
    // blockHandler：限流之后的自定义处理函数，需要和当前资源写在同一个类中，并且参数以及返回值也必须和资源的参数以及返回值是一致的，并且必须是public的。
    // fallback：熔断降级之后的自定义处理函数，需要和当前资源写在同一个类中，并且参数以及返回值也必须和资源的参数以及返回值是一致的，并且必须是public的。
    // 当blockHandler和fallback同时设置的时候，限流和熔断之后只会触发fallback。
    // 如果想将blockHandler和fallback的逻辑写在其他类中，还需要指定blockHandlerClass和fallbackClass，这种情况下，blockHandler和fallback的逻辑函数必须是static的。
    @SentinelResource(value = "domain", blockHandler = "blockHandler1",
            fallback = "fallback1")
    public String domain() {
        cloudAccoudService.getAccoud();
        return "你好，这是限流测试";
    }
	// 设置限流函数，函数名跟fallback设置的值一致
    public String blockHandler1() {
        System.out.println("这是限流");
        return "这是限流函数";
    }
	// 自定义熔断函数，函数名跟blockHandler设置的值一致
    public String fallback1() {
        System.out.println("这是熔断降级");
        return "这是熔断降级函数";
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

4、Sentinel限流源码解析

SentinelResourceAspect在SentinelAutoConfiguration中会被注册为Bean，是整个@SentinelResource的具体实现入口。

• SentinelResourceAspect

@Aspect
public class SentinelResourceAspect extends AbstractSentinelAspectSupport {
    public SentinelResourceAspect() {
    }

    @Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")
    public void sentinelResourceAnnotationPointcut() {
    }

    @Around("sentinelResourceAnnotationPointcut()")
    public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {
        Method originMethod = this.resolveMethod(pjp);
        SentinelResource annotation = (SentinelResource)originMethod.getAnnotation(SentinelResource.class);
	  ……省略部分代码……
                try {
                    // 就是最原始的Sentinel的实现
                    entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());
                    // Sentinel切面逻辑实现完成之后，执行具体的逻辑
                    Object result = pjp.proceed();
       ……省略部分代码……             
                
            } finally {
                if (entry != null) {
                    entry.exit(1, pjp.getArgs());
                }

            }
        }
    }
}
// SphU.entry(resourceName, resourceType, entryType, pjp.getArgs())方法最终会调用到CtSph.entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)这个方法中。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

• CtSph.entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object… args)

private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args) throws BlockException {
    Context context = ContextUtil.getContext();
    if (context instanceof NullContext) {
        return new CtEntry(resourceWrapper, (ProcessorSlot)null, context);
    } else {
             ……省略部分代码……         
        	// 以当前的资源名称resourceWrapper最为key，ProcessorSlotChain为valiue构造一个map集合。
            ProcessorSlot<Object> chain = this.lookProcessChain(resourceWrapper);
            if (chain == null) {
                return new CtEntry(resourceWrapper, (ProcessorSlot)null, context);
            } else {
                Entry e = new CtEntry(resourceWrapper, chain, context);

                try {
                    // 将链条中的每一个链都调用一遍
                    chain.entry(context, resourceWrapper, (Object)null, count, prioritized, args);
                } catch (BlockException var9) {
                    e.exit(count, args);
                    throw var9;
                } catch (Throwable var10) {
                    RecordLog.info("Sentinel unexpected exception", var10);
                }

                return e;
            }
        }
    }
}
// this.lookProcessChain(resourceWrapper);最终会调用到DefaultSlotChainBuilder.build()方法上来
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

• DefaultSlotChainBuilder.build()：实际上就是在构造执行链，将一系列的ProcessorSlot添加到ProcessorSlotChain链条当中。

public ProcessorSlotChain build() {
       // 创建一个DefaultProcessorSlotChain执行链条
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();
    	// 获取所有的ProcessorSlot具体实例。
        List<ProcessorSlot> sortedSlotList = SpiLoader.of(ProcessorSlot.class).loadInstanceListSorted();
        Iterator var3 = sortedSlotList.iterator();

        while(var3.hasNext()) {
            ……省略部分代码……    
            ProcessorSlot slot = (ProcessorSlot)var3.next();
            // 将所有的ProcessorSlot具体实例循环添加到ProcessorSlotChain执行链条当中
            chain.addLast((AbstractLinkedProcessorSlot)slot);
        }

        return chain;
    }
// 上述方法中最终会这些具体的执行链添加进DefaultProcessorSlotChain执行链条中。这些类在sentinel-core包下，采用SPI的方式被加载进去。
# Sentinel default ProcessorSlots
com.alibaba.csp.sentinel.slots.nodeselector.NodeSelectorSlot
com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot
com.alibaba.csp.sentinel.slots.logger.LogSlot
com.alibaba.csp.sentinel.slots.statistic.StatisticSlot
com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot
com.alibaba.csp.sentinel.slots.system.SystemSlot
com.alibaba.csp.sentinel.slots.block.flow.FlowSlot
com.alibaba.csp.sentinel.slots.block.degrade.DegradeSlot
// 最终的DefaultProcessorSlotChain执行链条形式如下：
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
 first -next-> NodeSelectorSlot -next-> …… DegradeSlot -next-> NULL		
 （头指针）													                   ↑ 
 							                                                                                                     																									next -> null
																								（尾指针）
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||     
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

• 统计规则中心调用链：StatisticSlot

用于存储资源的统计信息以及调用者信息，例如：该资源的RT，QPS，Thread count灯信息，这些信息将作为多维度限流，降级的依据，为后续的流控规则降级规则做准备。

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);
        // 当前正在调用当前资源的线程数加1
        node.increaseThreadNum();
         // 增加调用当前资源的校验规则通过的请求数
        node.addPassRequest(count);
        ……省略部分代码…… 
    } catch (BlockException var11) {
       ……省略部分代码…… 
    } catch (Throwable var12) {
        ……省略部分代码…… 
    }
   // 这两个catch块是实现整个限流熔断的关键，责任链上任何一个链执行出错的时候，就会被捕捉，从添加异常指标。
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

• 流控规则实现：FlowSlot

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
    // 先执行流控规则自己的校验
    this.checkFlow(resourceWrapper, context, node, count, prioritized);
    // 继续调用下一个链
    this.fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
// this.checkFlow(resourceWrapper, context, node, count, prioritized);最终会调用到FlowRuleChecker.checkFlow方法中。

1
2
3
4
5
6
7
8

• FlowRuleChecker.checkFlow

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) {
            // 从服务端获取所有的流控规则
            Collection<FlowRule> rules = (Collection)ruleProvider.apply(resource.getName());
            if (rules != null) {
                Iterator var8 = rules.iterator();
				// 循环所有的规则
                while(var8.hasNext()) {
                    FlowRule rule = (FlowRule)var8.next();
                    // 如果当前的资源的规则没有通过，那么即抛出FlowException，在StatisticSlot中有捕获异常，并且还会统计异常的次数。
                    if (!this.canPassCheck(rule, context, node, count, prioritized)) {
                        throw new FlowException(rule.getLimitApp(), rule);
                    }
                }
            }

        }
    }

// this.canPassCheck(rule, context, node, count, prioritized)会调用到passLocalCheck(rule, context, node, acquireCount, prioritized)这个方法，最终会调用到rule.getRater().canPass(selectedNode, acquireCount, prioritized)这个方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

• rule.getRater()

rule实际上就是FlowRule类型，而这个类中就定义了流控规则页面上所有的属性。

public class FlowRule extends AbstractRule {
    private int grade = 1; // 阈值类型：0-并发线程数，1-QPS
    private double count; // 单机阈值
    private int strategy = 0; // 流控模式：0-直接,1-关联，2-链路
    private String refResource; // 资源名
    private int controlBehavior = 0; // 流控效果：0-快速失败，1-Warm Up，2-排队等待
    private int warmUpPeriodSec = 10; // 
    private int maxQueueingTimeMs = 500; // 速率限制器行为中的最大排队时间。
    private boolean clusterMode; // 是否集群
    private ClusterFlowConfig clusterConfig;
    private TrafficShapingController controller;  
}
1
2
3
4
5
6
7
8
9
10
11
12

• rule.getRater().canPass

在canPass中针对配置的流控规则进行判断

// DefaultController：快速失败 ====》 滑动时间窗口
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
       // 根据阈值类型获取对应的指标，默认就是QPS
        int curCount = avgUsedTokens(node);
       // 当前着一秒内的QPS数 + 新过来的一个请求数 > 设置的阈值，如果检验不通过直接返回false，返回fasle，FlowRuleChecker.checkFlow中的canPassCheck返回fasle，取反就是true，直接抛出FlowException，然后被StatisticSlot捕捉，记录到异常请求数中。
        if (curCount + acquireCount > count) {
            ……省略部分代码……    
            return false;
        }
        return true;
    }

// 根据配置的阈值类型，获取对应的StatisticSlot统计出来的指标值。
private int avgUsedTokens(Node node) {
        if (node == null) {
            return DEFAULT_AVG_USED_TOKENS;
        }
    // 返回统计指标模式，是线程数呢还是QPS数。
        return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());
    }

// 在这里使用StatisticSlot统计出来的指标数据。
public double passQps() {
   return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}

// RateLimiterController：排队等待 ====》漏斗算法   
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
        // 如果没有请求，则通过。
        if (acquireCount <= 0) {
            return true;
        }
        // 单机阈值小于等于0，直接拒绝。
        if (count <= 0) {
            return false;
        }
		// 获取当前时间。
        long currentTime = TimeUtil.currentTimeMillis();
        // 计算两个请求之间的时间间隔，单位为ms，默认计算出来的就是1000ms。
        long costTime = Math.round(1.0 * (acquireCount) / count * 1000);

        // 允许第一个请求通过花费的时间，单位ms，latestPassedTime默认-1，也就是说，允许第一个请求通过的时间为999ms。
        long expectedTime = costTime + latestPassedTime.get();
		// 如果允许通过的时间小于等于当前时间，说明请求能在指定的时间内通过，这样当前请求处理成功。
        if (expectedTime <= currentTime) {
            // 记录本次请求通过的时间点。
            latestPassedTime.set(currentTime);
            return true;
        } else {
            // 如果允许通过的时间大于当前时间，说明当前请求需要被限流，至少要等到当前时间要大于等于等待时间。
            long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
            // 如果计算出来的当前的等待时间大于页面上设置的最大超时时间，那么当前请求执行失败。
            if (waitTime > maxQueueingTimeMs) {
                return false;
            } else {
                 // 如果计算出来的当前的等待时间小于页面上设置的最大超时时间
                // 计算出当前请求应该开始的时间，上一次请求结束的时间 + 请求之间的间隔时间就是下一次请求应该允许通过的开始时间。
                long oldTime = latestPassedTime.addAndGet(costTime);
                try {
                    // 重新计算下当前请求的等待时间
                    waitTime = oldTime - TimeUtil.currentTimeMillis();
                    // 如果计算出来的当前的等待时间大于页面上设置的最大超时时间，那么当前请求执行失败。
                    if (waitTime > maxQueueingTimeMs) {
                        // 上边计算当前请求应该开始的时间的时候，给上次请求结束时间增加了请求之间的间隔时间，请求失败的时候，需要将这个值还原。
                        latestPassedTime.addAndGet(-costTime);
                        return false;
                    }
                    // 经过上边的计算，如果计算出来的等待时间大于0，那么说明计算出当前请求应该开始的时间也是大于当前时间的，那么直接休眠当前线程一段时间，也就是当前线程需要排队等待，时间为计算出来的等待时间，休眠到之后，所有的程序也就回到正常情况下。
                    if (waitTime > 0) {
                        Thread.sleep(waitTime);
                    }
                    // 排队等待之后放行。
                    return true;
                } catch (InterruptedException e) {
                }
            }
        }
        return false;
    }    
    
//  WarmUpController：Warm Up  ===》令牌桶算法  
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
    // 获取当前时间窗口的qps
    long passQps = (long) node.passQps();
    // 获取上一个时间窗口的qps
    long previousQps = (long) node.previousPassQps();
    // 生成令牌，并控制令牌的填充和丢弃
    syncToken(previousQps);
	// 这里获取到当前令牌桶中的剩余的令牌数
    long restToken = storedTokens.get();
    // 如果剩余的令牌令牌数大于等于警戒值
    if (restToken >= warningToken) {
        // 获取超出警戒值的令牌数
        long aboveToken = restToken - warningToken;
        // 计算出产生令牌的速率
        double warningQps = Math.nextUp(1.0 / (aboveToken * slope + 1.0 / count));
        // 当前窗口的qps加上当前的请求数1，就是当前token的消费速率，如果消耗令牌的速度是远远是小于产生令牌的速率，也就说明当前有多余的令牌可供消耗，此时请求能通过，如果不是，那么就需要限流。
        if (passQps + acquireCount <= warningQps) {
            return true;
        }
    } else {
        // 如果剩余的令牌令牌数小于警戒值，但是当窗口的QPS + 当前的请求数1 <= 单机阈值的时候，此时请求也能通过，这是因为当前的令牌数虽然不处于饱和状态，但是当前的请求速率依旧处于阈值之内。如果不是，那么就需要限流。
        if (passQps + acquireCount <= count) {
            return true;
        }
    }
    return false;
}

// 生成令牌，并控制令牌的填充和丢弃
protected void syncToken(long passQps) {
    // 获取当前时间
    long currentTime = TimeUtil.currentTimeMillis();
    // 当前时间减去1s.
    currentTime = currentTime - currentTime % 1000;
    // 获取上次填充令牌的时间
    long oldLastFillTime = lastFilledTime.get();
    // 如果当前时间距离上次填充令牌的时间差1s,那就说明当前请求和上次请求在同一个时间段内，此时不需要操作令牌，避免重复操作令牌。
    // 如果当前时间小于上次填充令牌的时间，有可能是时钟出现了问题，那么此时也不能操作令牌。
    if (currentTime <= oldLastFillTime) {
        return;
    }
    // 获取目前剩余的令牌数。
    long oldValue = storedTokens.get();
    // 根据当前时间和上一个窗口的QPS计算出一个新的令牌桶中的令牌数量
    long newValue = coolDownTokens(currentTime, passQps);
    // 通过cas的方式，重置令牌数，这里可能直接将令牌数直接替换。
    if (storedTokens.compareAndSet(oldValue, newValue)) {
        // 重置令牌后的令牌数要减去上一个窗口的qps
        long currentValue = storedTokens.addAndGet(0 - passQps);
        // 如果减去当前的qps之后剩余的令牌数小于0，那就说明令牌桶中的令牌已经无法支撑当前这次请求了。
        if (currentValue < 0) {
            // 直接清空令牌空。
            storedTokens.set(0L);
        }
  		// 记录最后操作令牌的时间
      lastFilledTime.set(currentTime);
  }
}

// 根据当前时间和当前窗口的QPS计算出一个新的令牌桶中的令牌数量
private long coolDownTokens(long currentTime, long passQps) {
    // 获取当前令牌桶中的令牌数量
     long oldValue = storedTokens.get();
     // 记录当前令牌桶中的令牌数量
     long newValue = oldValue;

     // 添加令牌的判断前提条件:
     // 当前剩余令牌数小于警戒值的时候，也就是令牌的消耗程度远远高于警戒线的时候
     if (oldValue < warningToken) {
         // 计算出下一次可能处理的QPS，也就是需要增加令牌数.
         newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);
      // 当前剩余令牌数大于警戒值的时候，也就是令牌的消耗程度远远低于警戒线的时候
     } else if (oldValue > warningToken) {
         // 如果当前的QPS小于（count / coldFactor），就说明当前状态还是依旧处于令牌桶的容量调整阶段。
         if (passQps < (int)count / coldFactor) {
             // 计算出下一次可能处理的QPS，也就是需要增加令牌数.
             newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);
         }
     }
    // 如果产生令牌的数量多于令牌桶的最大容量，那么就会丢弃多余的令牌数，即只能保持令牌桶的最大容量。
     return Math.min(newValue, maxToken);
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

Sentinel滑动窗口实现总结：

在请求访问时,sentinel会创建一条调用链路:NodeSelectorSlot --> ClusterBuilderSlot --> LogSlot --> StatisticSlot ……，其中StatisticSlot是整个Sentinel实现的关键。
1、在调用到StatisticSlot的时候，传入DefaultNode，而DefaultNode继承于StatisticNode，StatisticNode会构建一个统计中心Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false)用来做统计。
2、StatisticSlot会先执行其他链条，然后再统计，如果请求成功就增加响应的线程数，通过的请求数，node.increaseThreadNum();node.addPassRequest(count);如果请求失败，那就增加记录响应的异常数，node.increaseBlockQps(count);。
3、根据当前请求时间，获取当前请求对应的时间窗口。首先根据当前时间计算出当前请求对应的时间窗口在数组中下标位置，在计算出当前时间在对应的时间窗口内的对应的起始位置。

• 假设根据下标没有找到对应的窗口，那么就new一个WindowWrap放到对应的位置上。
• 假设当前计算出来的窗口开始时间等于旧桶的开始时间戳，这意味着时间在桶内，所以直接返回桶。
• 假设当前计算出来的窗口开始时间大于旧桶的开始时间戳，这意味着当前这个桶不可用，必须将这个时间窗口中的是所有数据进行重置。比如将当前窗口内统计的所有的指标数据重置为0，将当前窗口的开始时间设置为计算出来的开始时间。

4、获取到对应的时间窗口的之后，会根据目前的执行情况将窗口中对应的指标增加，最终完成数据指标的统计。
5、流控规则（FlowSlot）和熔断规则（DegradeSlot）执行链条都会从数据统计中心rollingCounterInMinute中获取当前时间内的窗口的指标数据，然后再从服务端获取页面上配置的规则阈值进行判断，通过则增加通过请求数据node.increaseThreadNum();node.addPassRequest(count);，不通过则抛出异常，被StatisticSlot捕捉后，增加异常处理数，node.increaseBlockQps(count);。

Sentinel漏斗算法实现总结：

当页面上流控效果选择的是排队等待的时候，流控规则（FlowSlot）就会使用排队等待的算法（漏斗算法）。

1、根据单机阈值计算出请求之间的间隔时间，如果单机阈值为1，默认计算出来的间隔时间为1000ms。

2、计算每一个请求允许通过的时间区间，默认允许第一个请求通过的时间为999ms。

3、如果允许通过的时间区间小于等于当前的时间戳，那么请求通过。

4、如果允许通过的时间区间大于当前的时间戳，那么请求不允许通过，说明此次请求应该被限制访问。此时当前线程需要休眠，也就是排队等待，时间为计算出来的等待时间，休眠之后，允许被放行。

Sentinel令牌桶算法实现总结：

当页面上流控效果选择的是 Warm Up 的时候，流控规则（FlowSlot）就会使用令牌桶算法。

1、获取当前窗口和上一个窗口的QPS，然后根据上一个窗口的QPS调整令牌桶中的令牌。

• 如果当前请求时间小于或者等于上一次调整令牌桶的时间，这说明，说明当前请求和上次请求在同一个时间段内，此时不需要操作令牌，避免重复操作令牌。也有可能是时钟出现了问题，那么此时也不能操作令牌。
• 平衡令牌桶的令牌数：
  • 当剩余令牌数小于警戒值的时候，也即是消耗令牌的速率大于警戒值的时候，需要添加令牌。
  • 当剩余令牌数大于警戒值的时候，也即是目前正处于预热阶段，但是消耗令牌的速率小于系统最冷时的生成令牌的速率的时候，需要添加令牌。
  • 当生成的令牌数量大于令牌桶的容量的时候，需要丢弃多余的那些令牌。

2、判断是否限流：

• 如果剩余的令牌数大于等于警戒值，那么说明正在预热阶段，那么需要判断消费令牌的速率和生产令牌的速率，如果消费令牌的速率大于生产令牌的速率就放行，否则就限流。
• 如果剩余的令牌数小于警戒值，那么说明预热阶段已经结束，QPS平稳，那么需要判断消费令牌的速率和设置额的阈值，如果消费令牌的速率小于设置的阈值就放行，否则就限流。

5、Sentinel熔断降级源码解析

DegradeSlot在整个Sentinel中调用链中会被调用。

• DegradeSlot.entry

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

// 检查熔断状态
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)) {
            // 如果抛出异常，那么会在StatisticSlot被捕捉，触发熔断降级处理。
            throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule());
        }
    }
}

@Override
    public boolean tryPass(Context context) {
        // 如果熔断是关闭的，那么此时请求是允许通过的。
        if (currentState.get() == State.CLOSED) {
            return true;
        }
        // 如果熔断是打开的
        if (currentState.get() == State.OPEN) {
            // 但是当前请求的时间大于熔断时间，也就是熔断时间已经过了，那么此时就将熔断状态更改为半开状态，这种状态下，允许发送请求去探测目标服务是否已经恢复可用。
            return retryTimeoutArrived() && fromOpenToHalfOpen(context);
        }
        return false;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

• 切面类SentinelResourceAspect中最终在finally中调用了entry.exit(1, pjp.getArgs());

} finally {
    if (entry != null) {
        entry.exit(1, pjp.getArgs());
    }
}
1
2
3
4
5

• DegradeSlot.exit

// SentinelResourceAspect中调用的exit方法，最终会触发各个链条的exit方法。
@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) {
        // 这里请求就正常通过，并在CircuitBreaker中记录相应的指标。
        for (CircuitBreaker circuitBreaker : circuitBreakers) {
            // 具体跟踪记录相应的指标。
            circuitBreaker.onRequestComplete(context);
        }
    }

    fireExit(context, r, count, args);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

• ExceptionCircuitBreaker：异常比例

@Override
public void onRequestComplete(Context context) {
    Entry entry = context.getCurEntry();
    if (entry == null) {
        return;
    }
    // 这里获取的是业务上的异常
    Throwable error = entry.getError();
    // 获取当前时间窗口的计数器
    SimpleErrorCounter counter = stat.currentWindow().value();
    if (error != null) {
        // 如果出现异常，异常数加1
        counter.getErrorCount().add(1);
    }
    // 总共请求书加1
    counter.getTotalCount().add(1);

    // 根据指标改变熔断的状态
    handleStateChangeWhenThresholdExceeded(error);
}

private void handleStateChangeWhenThresholdExceeded(Throwable error) {
    // 如果熔断器是打开状态的，直接返回，那么此次请求不允许请求远程服务,直接响应熔断
    if (currentState.get() == State.OPEN) {
        return;
    }
    // 如果熔断器是半打开状态的
    if (currentState.get() == State.HALF_OPEN) {
        // 如果请求远程服务没有发生异常，
        if (error == null) {
            // 那么将熔断器从半打开状态更改为关闭状态，因为此时请求成功的。
            fromHalfOpenToClose();
        } else {
            // 如果请求远程服务发生了异常，那么将熔断器从半打开状态更改为打开状态，因为此时请求不成功，发生了异常。
            fromHalfOpenToOpen(1.0d);
        }
        return;
    }
    // 获取所有的计数器
    List<SimpleErrorCounter> counters = stat.values();
    long errCount = 0;
    long totalCount = 0;
    // 循环计算整体的错误数和请求总数。
    for (SimpleErrorCounter counter : counters) {
        errCount += counter.errorCount.sum();
        totalCount += counter.totalCount.sum();
    }
    // 如果整体请求总数 < 最小请求数，此时不做熔断处理,不用更改熔断器的状态。
    if (totalCount < minRequestAmount) {
        return;
    }
    double curCount = errCount;
    // 如果熔断策略使用的异常比例
    if (strategy == DEGRADE_GRADE_EXCEPTION_RATIO) {
        // 计算异常比例
        curCount = errCount * 1.0d / totalCount;
    }
    // 如果异常比例大于设置的异常比例阈值，那么熔断器打开
    if (curCount > threshold) {
        transformToOpen(curCount);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

• ResponseTimeCircuitBreaker：慢调用比例。

public void onRequestComplete(Context context) {
    // 获取当前的时间窗口的计数器
    SlowRequestCounter counter = slidingCounter.currentWindow().value();
    Entry entry = context.getCurEntry();
    if (entry == null) {
        return;
    }
    long completeTime = entry.getCompleteTimestamp();
    if (completeTime <= 0) {
        completeTime = TimeUtil.currentTimeMillis();
    }
    // 计算当前请求最大响应时长（RT）
    long rt = completeTime - entry.getCreateTimestamp();
    // 如果当前请求最大响应时长大于设置的请求最大响应时长阈值
    if (rt > maxAllowedRt) {
        // 慢调用计数加1
        counter.slowCount.add(1);
    }
    // 请求总数加1
    counter.totalCount.add(1);
	// 根据指标改变熔断的状态
    handleStateChangeWhenThresholdExceeded(rt);
}

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;
    }

    // 获取所有的计数器
    List<SlowRequestCounter> counters = slidingCounter.values();
    long slowCount = 0;
    long totalCount = 0;
    // 循环计算整体的慢调用请求数和请求总数。
    for (SlowRequestCounter counter : counters) {
        slowCount += counter.slowCount.sum();
        totalCount += counter.totalCount.sum();
    }
    // 如果整体请求总数 < 最小请求数，此时不做熔断处理,不用更改熔断器的状态。
    if (totalCount < minRequestAmount) {
        return;
    }
    // 计算当前的慢调用比例
    double currentRatio = slowCount * 1.0d / totalCount;
    // 如果慢调用比例大于设置的慢调用比例阈值，
    if (currentRatio > maxSlowRequestRatio) {
        // 将熔断器的状态更改为打开状态，
        transformToOpen(currentRatio);
    }
    // 如果设置的慢调用比例和设置的慢调用比例阈值相等，并且都为1的时候，更改熔断器的状态为打开状态。
    if (Double.compare(currentRatio, maxSlowRequestRatio) == 0 &&
            Double.compare(maxSlowRequestRatio, SLOW_REQUEST_RATIO_MAX_VALUE) == 0) {
        transformToOpen(currentRatio);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68