流控神器Sentinel指南：深入分析流控原理（下）

2024-03-28 15:38
文章标签 原理 sentinel 指南 深入分析 神器 流控

本文主要是介绍流控神器Sentinel指南：深入分析流控原理（下），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

        • 1. 数据统计中心
        • 2. 滑动窗口
          • 2.1 ArrayMetric
          • 2.2 LeapArray
          • 2.3 统计核心
        • 3. 三种流控方式
          • 3.1 Default
          • 3.2 Warm Up
          • 3.3 匀速器
        • 4. 退出的流程
          • 4.1 StatisticSlot#exit
          • 4.2 ContextUtil#exit
        • 下期预告：

上篇文章中从规则的定义开始说起，再到调用树，Context、slotChain等，详细的介绍了访问资源，到FlowSlot的entry方法为止。
本文将继续上篇文章的主线，阐述流量控制的实现原理。在介绍之前我们需要对滑动窗口实时指标统计进行分析。

1. 数据统计中心

上篇文章中提到过StatisticSlot的作用是统计资源的实时数据，请求经过这里，然后FlowSlot根据规则进行匹配对比，决定是放行还是限制。当返回之后，StatisticSlot统计请求的信息。当然逻辑的实现是在entry方法中。

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, Object... args)throws Throwable {try {// 直接触发下游slot的entry操作fireEntry(context, resourceWrapper, node, count, args);// 如果到达这里说明获取token成功//增加并发线程数，node是DefaultNodenode.increaseThreadNum();// 增加每秒的请求数node.addPassRequest();if (context.getCurEntry().getOriginNode() != null) {// 如果OriginNod（调用方）不为空，同样增加线程数和请求数，没有调用ContextUtil.enter，OriginNode为空context.getCurEntry().getOriginNode().increaseThreadNum();context.getCurEntry().getOriginNode().addPassRequest();}// 这里是全局的统计if (resourceWrapper.getType() == EntryType.IN) {Constants.ENTRY_NODE.increaseThreadNum();Constants.ENTRY_NODE.addPassRequest();}// 关于做参数限流的，暂时不考虑for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {handler.onPass(context, resourceWrapper, node, count, args);}} catch (BlockException e) {context.getCurEntry().setError(e);// 增加阻塞数node.increaseBlockedQps();if (context.getCurEntry().getOriginNode() != null) {// 增加调用方node的阻塞数context.getCurEntry().getOriginNode().increaseBlockedQps();}if (resourceWrapper.getType() == EntryType.IN) {// 全局阻塞数Constants.ENTRY_NODE.increaseBlockedQps();}for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {handler.onBlocked(e, context, resourceWrapper, node, count, args);}throw e;}
}

StatisticSlot的实现比较直观，触发下游的Slot entry方法，根据是否抛出BlockException来进行指标统计。似乎更神秘的逻辑是在Node中的。
进入到DefaultNode的addPassRequest:

@Override
public void addPassRequest() {super.addPassRequest();this.clusterNode.addPassRequest();
}

这里做了两件事，增加自身保存请求数。增加关联的clusterNode的请求数，这里就体现出clusterNode保存一个资源的指标数据作用了。注意clusterNode 是StatisticNode 类型的，因此不会存在无限递归的情况。父类StatisticNode addPassRequest的方法如下：

@Override
public void addPassRequest() {rollingCounterInSecond.addPass();rollingCounterInMinute.addPass();
}

Metric rollingCounterInSecond是统计一秒内的指标数据数据的，rollingCounterInMinute是统计一分钟的数据。这两个成员变量在StatisticNode定义如下：

private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,IntervalProperty.INTERVAL);
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);

下面结合滑动窗口来深入分析这两个数据结构。

2. 滑动窗口
2.1 ArrayMetric

在StatisticNode中有两个ArrayMetric类型的属性。ArrayMetric实现了Metric接口，Metric是度量指标的抽象。ArrayMetric是它的实现类。它相当于是对底层指标的代理封装。ArrayMetric的构造方法

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

其中OccupiableBucketLeapArray和BucketLeapArray数据结构保存了实时的指标数据，简单的说是一个数组,数组元素是WindowWrap(窗口桶)。这两个类都继承自LeapArray类。

2.2 LeapArray

LeapArray为保存数据的基类，先看它的构造方法：

//LeapArray#LeapArray
public LeapArray(int sampleCount, int intervalInMs) {// 单个窗口桶的时间长度(以毫秒为单位)this.windowLengthInMs = intervalInMs / sampleCount;// LeapArray的总时间跨度，以秒为单位this.intervalInMs = intervalInMs;// 数组长度，即窗口的个数this.sampleCount = sampleCount;// 数组元素为WindowWrap，WindowWrap保存了MetricBucket，在它内部才保存真正的指标数据this.array = new AtomicReferenceArray<WindowWrap<T>>(sampleCount);
}

WindowWrap封装了表示一个时间窗口，它有三个重要的属性：

public class WindowWrap<T> {// 窗口的长度private final long windowLengthInMs;// 起始毫秒数private long windowStart;// 指标数据，一般是MetricBucket,private T value;
}

MetricBucket代表了在某一个时间窗口内的所有数据，如RT、Success、Pass、Block等等。

public class MetricBucket {private final LongAdder[] counters;
}

这里用到LongAdder,JDK1.8中新添加的一种原子类，对于数据统计场景比Atomic原子类具有更好的并发性。所有的数据都是保存在LongAdder数组中。

2.3 统计核心

回到rollingCounterInSecond.addPass()方法

@Override
public void addPass() {// 获取当前窗口WindowWrap<MetricBucket> wrap = data.currentWindow();//对MetricBucket +1 操作wrap.value().addPass();
}

继续跟进currentWindow方法：

public WindowWrap<T> currentWindow(long time) {// 到目前为止总共窗口的数量long timeId = time / windowLengthInMs;// 当前时间滑动出口数组的下标int idx = (int)(timeId % array.length());// 计算出当前窗口的起始时间time = time - time % windowLengthInMs;while (true) {WindowWrap<T> old = array.get(idx);if (old == null) {// 如果获取到的WindowWrap为空，则新创建一个，第三个参数是创建一个空的BucketWindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, time, newEmptyBucket());if (array.compareAndSet(idx, null, window)) // 将WindowWrap置入数组中return window;} else {// 设置失败，当前线程让出时间片等待Thread.yield();}// 如果当前窗口的开始时间与old的开始时间相等，则直接返回old窗口} else if (time == old.windowStart()) {return old;// 如果当前时间窗口的开始时间已经超过了old窗口的开始时间，则放弃old窗口// 并将time设置为新的时间窗口的开始时间，此时窗口向前滑动} else if (time > old.windowStart()) {if (updateLock.tryLock()) {try {return resetWindowTo(old, time);} finally {updateLock.unlock();}} else {Thread.yield();}// 这个条件不可能存在} else if (time < old.windowStart()) {// Cannot go through here.return new WindowWrap<T>(windowLengthInMs, time, newEmptyBucket());}}
}

过程有些复杂，终极技能看图：
sentinel
初始的时候arrays数组中只有两个窗口，每个时间窗口的长度是500ms，当前时间走到1000或者1499时，时间窗口是第一个，同理时间走到1500至1999时，是第二个窗口。当时间主线走到2000时，会替换数组第一个start time,并且初始化指标数据(置0),当时间主线走到2500时，会替换数组第二个start time，以此一直循环。
这个数组就像一个窗口在时间主线上滑动。从而我们可以获取到每秒的指标数据。

回到rollingCounterInSecond.addPass()，当我们获取到了当前的滑动窗口，对它持有的MetricBucket进行加1操作。

3. 三种流控方式

继续上篇文章结束的地方，FlotSlot通过三种策略进行流量控制，分别为Default、warm up、rate limiter。在创建规则时通过不同的设置创建对应的Controller。
sentinel
下面逐个分析。

3.1 Default

默认的流控方式，如果达到阈值直接拒绝请求。下面是它的实现逻辑。

 @Override
public boolean canPass(Node node, int acquireCount) {// 获取当前请求数量int curCount = avgUsedTokens(node);// 如果当前数量加上本次请求的个数大于阈值，返回falseif (curCount + acquireCount > count) {return false;}return true;
}private int avgUsedTokens(Node node) {if (node == null) {return -1;}// 如果根据QPS进行限流，调用clusterNode的passQpsreturn grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)node.passQps();
}

继续进入passQps方法，实现在StatisticNode类中。

public long passQps() {// INTERVAL为1return rollingCounterInSecond.pass() / IntervalProperty.INTERVAL;
}

rollingCounterInSecond 在前面说过了，它持有滑动窗口leapArray的引用。保存了一秒内的指标数据。数组长度为2。对StatisticNode暴露出查询指标数据的接口。继续跟进pass方法

@Override
public long pass() {data.currentWindow();// 这里的作用是刷新窗口long pass = 0;// 获取数组中所有WindoWrap关联的MetricBucketList<MetricBucket> list = data.values();// 遍历所有MetricBucket，获取pass数量for (MetricBucket window : list) {pass += window.pass();}return pass;
}

整体流程比较直观简单，注意一个细节就是代码第一行，作用是刷新当前窗口，避免获取过期的数据。

3.2 Warm Up
3.3 匀速器

匀速器模式是让所有的请求匀速的通过，请求进入后可能需要等待，因此如果等待时间超过设置的阈值，那么该请求以拒绝而结束。看它的实现

    @Overridepublic boolean canPass(Node node, int acquireCount) {long currentTime = TimeUtil.currentTimeMillis();// count即设置规则时QPS数,costTime 指获取信号所需要的时间long costTime = Math.round(1.0 * (acquireCount) / count* 1000);//预期通过的时间，latestPassedTime是前一个请求通过的时间long expectedTime = costTime + latestPassedTime.get();// 第一个请求必定进入该分支，还有一种情况 当count值设置的过大，导致costTime 为0，那么后面的// 绝大部分请求都会进入此分支if (expectedTime <= currentTime) {latestPassedTime.set(currentTime);return true;} else {// 计算需要的等待时间,即costTime + 上一个请求的通过的时间 - 当前时间long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();// 如果等待时间大于阈值，pass 失败if (waitTime >= maxQueueingTimeMs) {return false;} else {// 尝试通过，设置上一次请求通过时间long oldTime = latestPassedTime.addAndGet(costTime);try {// 在次获取等待时间waitTime = oldTime - TimeUtil.currentTimeMillis();if (waitTime >= maxQueueingTimeMs) {latestPassedTime.addAndGet(-costTime);return false;}Thread.sleep(waitTime);return true;} catch (InterruptedException e) {}}}return false;}

整体过程比较清晰，但是存在一个问题，就是当count值设置过大，或者maxQueueingTimeMs设置过小可能会导致匀速器失效。建议count设置小于1000，maxQueueingTimeMs大于1。

4. 退出的流程

一般在入口处的finally语句块中会有entry.exit的调用，该方法会调用Entry的tureExit方法：

@Override
protected Entry trueExit(int count, Object... args) throws ErrorEntryFreeException {exitForContext(context, count, args);return parent;
}

显然更多的逻辑是在exitForContext方法的内部

protected void exitForContext(Context context, int count, Object... args) throws ErrorEntryFreeException {if (context != null) {if (context instanceof NullContext) {return;}if (context.getCurEntry() != this) {String curEntryNameInContext = context.getCurEntry() == null ? null : context.getCurEntry().getResourceWrapper().getName();// Clean previous call stack.CtEntry e = (CtEntry)context.getCurEntry();while (e != null) {e.exit(count, args);e = (CtEntry)e.parent;}String errorMessage = String.format("The order of entry exit can't be paired with the order of entry"+ ", current entry in context: <%s>, but expected: <%s>", curEntryNameInContext, resourceWrapper.getName());throw new ErrorEntryFreeException(errorMessage);} else {if (chain != null) {chain.exit(context, resourceWrapper, count, args);}// 设置CurEntry为Eenry的父亲context.setCurEntry(parent);if (parent != null) {((CtEntry)parent).child = null;// 将父Entry节点的child 置为null}if (parent == null) {//当前Entry的父Entry为null时，此时说明该Entry已经是最顶层的根节点了，可以清除context（ThreadLocal中）if (ContextUtil.isDefaultContext(context)) {ContextUtil.exit();}}// Clean the reference of context in current entry to avoid duplicate exit.clearEntryContext();}}
}

这里有两个重点：

  1. 通知slotChain，Entry退出了，其中重点是在StatisticSlot#exit方法中。
  2. 如果当前Entry的父Entry为null时，删除context

下面逐个分析

4.1 StatisticSlot#exit

StatisticSlot#exit中主要是对正常通过的请求数据进行一个统计，没能成功的请求在StatisticSlot#entry方法中的catch块中已经进行统计。所以这里是对通过的请求做一个指标统计。

 public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {DefaultNode node = (DefaultNode)context.getCurNode();if (context.getCurEntry().getError() == null) {long rt = TimeUtil.currentTimeMillis() - context.getCurEntry().getCreateTime();if (rt > Constants.TIME_DROP_VALVE) {rt = Constants.TIME_DROP_VALVE;}// 统计rtnode.rt(rt);if (context.getCurEntry().getOriginNode() != null) {context.getCurEntry().getOriginNode().rt(rt);}// 减少线程数node.decreaseThreadNum();if (context.getCurEntry().getOriginNode() != null) {context.getCurEntry().getOriginNode().decreaseThreadNum();}// 全局统计if (resourceWrapper.getType() == EntryType.IN) {Constants.ENTRY_NODE.rt(rt);Constants.ENTRY_NODE.decreaseThreadNum();}} else {// Error may happen.}Collection<ProcessorSlotExitCallback> exitCallbacks = StatisticSlotCallbackRegistry.getExitCallbacks();for (ProcessorSlotExitCallback handler : exitCallbacks) {handler.onExit(context, resourceWrapper, count, args);}fireExit(context, resourceWrapper, count);}
4.2 ContextUtil#exit

ContextUtil#exit很明确，从ThreadLocal中将Context置为null

public static void exit() {Context context = contextHolder.get();if (context != null && context.getCurEntry() == null) {contextHolder.set(null);}
}
下期预告：

熔断降级源码分析
Sentinel基于Netty的实时数据的传输实现细节

这篇关于流控神器Sentinel指南：深入分析流控原理（下）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/856013。

相关文章

Android开发环境配置避坑指南

Android开发环境配置避坑指南

《Android开发环境配置避坑指南》本文主要介绍了Android开发环境配置过程中遇到的问题及解决方案,包括VPN注意事项、工具版本统一、Gerrit邮箱配置、Git拉取和提交代码、MergevsR... 目录网络环境：VPN 注意事项工具版本统一：android Studio & JDKGerrit的邮
阅读更多...
基于Python打造一个智能单词管理神器

基于Python打造一个智能单词管理神器

《基于Python打造一个智能单词管理神器》这篇文章主要为大家详细介绍了如何使用Python打造一个智能单词管理神器,从查询到导出的一站式解决,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1. 项目概述：为什么需要这个工具2. 环境搭建与快速入门2.1 环境要求2.2 首次运行配置3. 核心功能使用指
阅读更多...
redis中使用lua脚本的原理与基本使用详解

redis中使用lua脚本的原理与基本使用详解

《redis中使用lua脚本的原理与基本使用详解》在Redis中使用Lua脚本可以实现原子性操作、减少网络开销以及提高执行效率,下面小编就来和大家详细介绍一下在redis中使用lua脚本的原理... 目录Redis 执行 Lua 脚本的原理基本使用方法使用EVAL命令执行 Lua 脚本使用EVALSHA命令
阅读更多...
SpringBoot整合OpenFeign的完整指南

SpringBoot整合OpenFeign的完整指南

《SpringBoot整合OpenFeign的完整指南》OpenFeign是由Netflix开发的一个声明式Web服务客户端,它使得编写HTTP客户端变得更加简单,本文为大家介绍了SpringBoot... 目录什么是OpenFeign环境准备创建 Spring Boot 项目添加依赖启用 OpenFeig
阅读更多...
Java Spring 中 @PostConstruct 注解使用原理及常见场景

Java Spring 中 @PostConstruct 注解使用原理及常见场景

《JavaSpring中@PostConstruct注解使用原理及常见场景》在JavaSpring中,@PostConstruct注解是一个非常实用的功能,它允许开发者在Spring容器完全初... 目录一、@PostConstruct 注解概述二、@PostConstruct 注解的基本使用2.1 基本代
阅读更多...
Golang HashMap实现原理解析

Golang HashMap实现原理解析

《GolangHashMap实现原理解析》HashMap是一种基于哈希表实现的键值对存储结构,它通过哈希函数将键映射到数组的索引位置,支持高效的插入、查找和删除操作,：本文主要介绍GolangH... 目录HashMap是一种基于哈希表实现的键值对存储结构，它通过哈希函数将键映射到数组的索引位置，支持
阅读更多...
SpringBoot请求参数接收控制指南分享

SpringBoot请求参数接收控制指南分享

《SpringBoot请求参数接收控制指南分享》：本文主要介绍SpringBoot请求参数接收控制指南,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Spring Boot 请求参数接收控制指南1. 概述2. 有注解时参数接收方式对比3. 无注解时接收参数默认位置
阅读更多...
CentOS7更改默认SSH端口与配置指南

CentOS7更改默认SSH端口与配置指南

《CentOS7更改默认SSH端口与配置指南》SSH是Linux服务器远程管理的核心工具,其默认监听端口为22,由于端口22众所周知,这也使得服务器容易受到自动化扫描和暴力破解攻击,本文将系统性地介绍... 目录引言为什么要更改 SSH 默认端口？步骤详解：如何更改 Centos 7 的 SSH 默认端口1
阅读更多...
SpringBoot多数据源配置完整指南

SpringBoot多数据源配置完整指南

《SpringBoot多数据源配置完整指南》在复杂的企业应用中,经常需要连接多个数据库,SpringBoot提供了灵活的多数据源配置方式,以下是详细的实现方案,需要的朋友可以参考下... 目录一、基础多数据源配置1. 添加依赖2. 配置多个数据源3. 配置数据源Bean二、JPA多数据源配置1. 配置主数据
阅读更多...
python中各种常见文件的读写操作与类型转换详细指南

python中各种常见文件的读写操作与类型转换详细指南

《python中各种常见文件的读写操作与类型转换详细指南》这篇文章主要为大家详细介绍了python中各种常见文件（txt,xls,csv,sql,二进制文件）的读写操作与类型转换,感兴趣的小伙伴可以跟... 目录1.文件txt读写标准用法1.1写入文件1.2读取文件2. 二进制文件读取3. 大文件读取3.1
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2