一、从一次 HTTP 请求开始
在一个生产环境中，服务节点通常暴露了成百上千个 HTTP 接口对外提供服务。为了保证系统的稳定性，核心 HTTP 接口往往需要配置限流规则。给 HTTP 接口配置限流，可以防止突发或恶意的高并发请求耗尽服务器资源（如 CPU、内存、数据库连接等），从而避免服务崩溃或引发雪崩效应。
基础示例
假设我们有下面这样一个 HTTP 接口，需要给它配置限流规则：
@RestController
@RequiredArgsConstructor
@RequestMapping("/demo")
public class DemoController {
@RequestMapping("/hello")
@SentinelResource("test_sentinel")
public String hello() {
return "hello world";
}
}
使用起来非常简单。首先我们可以选择给接口加上 @SentinelResource 注解（也可以不加，如果不加 Sentinel 客户端会使用请求路径作为资源名，详细原理在后面章节讲解），然后到流控控制台给该资源配置流控规则即可。
二、限流规则的加载
限流规则的生效，是从限流规则的加载开始的。聚焦到客户端的 RuleLoader 类，可以看到它支持了多种规则的加载：
- 流控规则；
- 集群限流规则；
- 熔断规则；
- ......
RuleLoader 核心逻辑
RuleLoader 类的核心作用是将这些规则加载到缓存中，方便后续使用：
public class RuleLoader {
/**
* 加载所有 Sentinel 规则到内存缓存
*
* @param sentinelRules 包含各种规则的配置对象
*/
public static void loadRule(SentinelRules sentinelRules) {
if (sentinelRules == null) {
return;
}
// 加载流控规则
FlowRuleManager.loadRules(sentinelRules.getFlowRules());
// 加载集群流控规则
RuleManager.loadClusterFlowRule(sentinelRules.getFlowRules());
// 加载参数流控规则
ParamFlowRuleManager.loadRules(sentinelRules.getParamFlowRules());
// 加载参数集群流控规则
RuleManager.loadClusterParamFlowRule(sentinelRules.getParamFlowRules());
// 加载熔断规则
DegradeRuleManager.loadRules(sentinelRules.getDegradeRules());
// 加载参数熔断规则
ParamDegradeRuleManager.loadRules(sentinelRules.getParamDegradeRules());
// 加载系统限流规则
SystemRuleManager.loadRules(sentinelRules.getSystemRules());
}
}
流控规则加载详情
以流控规则的加载为例深入FlowRuleManager.loadRules 方法可以看到其完整的加载逻辑：
public static void loadRules(List<FlowRule> rules) {
// 通过动态配置属性更新规则值
currentProperty.updateValue(rules);
}
updateValue 方法负责通知所有监听器配置变更：
public boolean updateValue(T newValue) {
// 如果新旧值相同，无需更新
if (isEqual(value, newValue)) {
return false;
}
RecordLog.info("[DynamicSentinelProperty] Config will be updated to: " + newValue);
// 更新配置值
value = newValue;
// 通知所有监听器配置已更新
for (PropertyListener<T> listener : listeners) {
listener.configUpdate(newValue);
}
return true;
}
FlowPropertyListener 是流控规则变更的具体监听器实现：
private static final class FlowPropertyListener implements PropertyListener<List<FlowRule>> {
@Override
public void configUpdate(List<FlowRule> value) {
// 构建流控规则映射表（按资源名分组）
Map<String, List<FlowRule>> rules = FlowRuleUtil.buildFlowRuleMap(value);
if (rules != null) {
// 清空旧规则
flowRules.clear();
// 加载新规则
flowRules.putAll(rules);
}
RecordLog.info("[FlowRuleManager] Flow rules received: " + flowRules);
}
}
三、SentinelServletFilter 过滤器
在 Sentinel 中，所有的资源都对应一个资源名称和一个 Entry。Entry 可以通过对主流框架的适配自动创建，也可以通过注解的方式或调用 API 显式创建。Entry 是限流的入口类，通过 @SentinelResource 注解的限流本质上也是通过 AOP 的方式进行了对 Entry 类的调用。
Entry 的编程范式
Entry 类的标准使用方式如下：
// 资源名可使用任意有业务语义的字符串，比如方法名、接口名或其它可唯一标识的字符串
try (Entry entry = SphU.entry("resourceName")) {
// 被保护的业务逻辑
// do something here...
} catch (BlockException ex) {
// 资源访问阻止，被限流或被降级
// 在此处进行相应的处理操作
}
Servlet Filter 拦截逻辑
对于一个 HTTP 资源，在没有显式标注 @SentinelResource 注解的情况下，会有一个 Servlet Filter 类 SentinelServletFilter 统一进行拦截：
public class SentinelServletFilter implements Filter {
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
throws IOException, ServletException {
HttpServletRequest sRequest = (HttpServletRequest) request;
Entry urlEntry = null;
try {
// 获取并清理请求路径
String target = FilterUtil.filterTarget(sRequest);
// 统一 URL 清理逻辑
// 对于 RESTful API，必须对 URL 进行清理（例如将 /foo/1 和 /foo/2 统一为 /foo/:id），
// 否则上下文和资源的数量会超过阈值
SentinelUrlCleaner urlCleaner = SentinelUrlCleaner.SENTINEL_URL_CLEANER;
if (urlCleaner != null) {
target = urlCleaner.clean(sRequest, target);
}
// 如果请求路径不为空且非安全扫描，则进入限流逻辑
if (!StringUtil.isEmpty(target) && !isSecScan) {
// 解析来源标识（用于来源限流）
String origin = parseOrigin(sRequest);
// 确定上下文名称
String contextName = webContextUnify
? WebServletConfig.WEB_SERVLET_CONTEXT_NAME
: target;
// 使用 WEB_SERVLET_CONTEXT_NAME 作为当前 Context 的名字
ContextUtil.enter(contextName, origin);
// 根据配置决定是否包含 HTTP 方法
if (httpMethodSpecify) {
String pathWithHttpMethod = sRequest.getMethod().toUpperCase() + COLON + target;
// 实际进入到限流统计判断逻辑，资源名是 "方法:路径"
urlEntry = SphU.entry(pathWithHttpMethod, ResourceTypeConstants.COMMON_WEB, EntryType.IN);
} else {
// 实际进入到限流统计判断逻辑，资源名是请求路径
urlEntry = SphU.entry(target, ResourceTypeConstants.COMMON_WEB, EntryType.IN);
}
}
// 继续执行后续过滤器
chain.doFilter(request, response);
} catch (BlockException e) {
// 处理被限流的情况
HttpServletResponse sResponse = (HttpServletResponse) response;
// 返回限流页面或重定向到其他 URL
WebCallbackManager.getUrlBlockHandler().blocked(sRequest, sResponse, e);
} catch (IOException | ServletException | RuntimeException e2) {
// 记录异常信息用于统计
Tracer.traceEntry(e2, urlEntry);
throw e2;
} finally {
// 释放 Entry 资源
if (urlEntry != null) {
urlEntry.exit();
}
// 退出当前上下文
ContextUtil.exit();
}
}
}
四、SentinelResourceAspect 切面
如果在接口上标注了 @SentinelResource 注解，还会有另外的逻辑处理。Sentinel 定义了一个单独的 AOP 切面 SentinelResourceAspect 专门用于处理注解限流。
SentinelResource 注解定义
先来看看 @SentinelResource 注解的完整定义：
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface SentinelResource {
/**
* Sentinel 资源的名称（即资源标识）
* 必填项，不能为空
*/
String value() default "";
/**
* 资源的入口类型（入站 IN 或出站 OUT）
* 默认为出站（OUT）
*/
EntryType entryType() default EntryType.OUT;
/**
* 资源的分类（类型）
* 自 1.7.0 版本起支持
*/
int resourceType() default 0;
/**
* 限流或熔断时调用的 block 异常处理方法的名称
* 默认为空（即不指定）
*/
String blockHandler() default "";
/**
* blockHandler 所在的类
* 如果与原方法不在同一个类，需要指定此参数
*/
Class<?>[] blockHandlerClass() default {};
/**
* 降级（fallback）方法的名称
* 默认为空（即不指定）
*/
String fallback() default "";
/**
* 用作通用的默认降级方法
* 该方法不能接收任何参数，且返回类型需与原方法兼容
*/
String defaultFallback() default "";
/**
* fallback 所在的类
* 如果与原方法不在同一个类，需要指定此参数
*/
Class<?>[] fallbackClass() default {};
/**
* 需要被追踪并触发 fallback 的异常类型列表
* 默认为 Throwable（即所有异常都会触发 fallback）
*/
Class<? extends Throwable>[] exceptionsToTrace() default {Throwable.class};
/**
* 指定需要忽略的异常类型（即这些异常不会触发 fallback）
* 注意：exceptionsToTrace 和 exceptionsToIgnore 不应同时使用；
* 若同时存在，exceptionsToIgnore 优先级更高
*/
Class<? extends Throwable>[] exceptionsToIgnore() default {};
}
实际使用示例
下面是一个完整的使用示例，展示了 @SentinelResource 注解的各种配置方式：
@RestController
public class SentinelController {
@Autowired
private ISentinelService service;
@GetMapping(value = "/hello/{s}")
public String apiHello(@PathVariable long s) {
return service.hello(s);
}
}
public interface ISentinelService {
String hello(long s);
}
@Service
@Slf4j
public class SentinelServiceImpl implements ISentinelService {
/**
* Sentinel 提供了 @SentinelResource 注解用于定义资源
*
* @param s 输入参数
* @return 返回结果
*/
@Override
// value：资源名称，必需项（不能为空）
// blockHandler：对应处理 BlockException 的函数名称
// fallback：用于在抛出异常的时候提供 fallback 处理逻辑
@SentinelResource(value = "hello", blockHandler = "exceptionHandler", fallback = "helloFallback")
public String hello(long s) {
log.error("hello:{}", s);
return String.format("Hello at %d", s);
}
/**
* Fallback 函数
* 函数签名与原函数一致，或加一个 Throwable 类型的参数
*/
public String helloFallback(long s) {
log.error("helloFallback:{}", s);
return String.format("Halooooo %d", s);
}
/**
* Block 异常处理函数
* 参数最后多一个 BlockException，其余与原函数一致
*/
public String exceptionHandler(long s, BlockException ex) {
// Do some log here.
log.error("exceptionHandler:{}", s);
ex.printStackTrace();
return "Oops, error occurred at " + s;
}
}
SentinelResourceAspect 核心逻辑
@SentinelResource 注解由 SentinelResourceAspect 切面处理，核心逻辑如下：
@Aspect
public class SentinelResourceAspect extends AbstractSentinelAspectSupport {
@Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")
public void sentinelResourceAnnotationPointcut() {
}
@Around("sentinelResourceAnnotationPointcut()")
public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {
// 获取目标方法
Method originMethod = resolveMethod(pjp);
// 获取注解信息
SentinelResource annotation = originMethod.getAnnotation(SentinelResource.class);
if (annotation == null) {
throw new IllegalStateException("Wrong state for SentinelResource annotation");
}
// 获取资源配置信息
String resourceName = getResourceName(annotation.value(), originMethod);
EntryType entryType = annotation.entryType();
int resourceType = annotation.resourceType();
Entry entry = null;
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();
// 优先检查忽略列表
if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {
throw ex;
}
// 检查异常是否在追踪列表中
if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {
traceException(ex);
// 执行 fallback 逻辑
return handleFallback(pjp, annotation, ex);
}
// 没有 fallback 函数可以处理该异常，直接抛出
throw ex;
} finally {
// 释放 Entry 资源
if (entry != null) {
entry.exit(1, pjp.getArgs());
}
}
}
/**
* 处理 BlockException
*
* blockHandler / blockHandlerClass 说明：
* - blockHandler：对应处理 BlockException 的函数名称，可选项
* - blockHandler 函数签名：与原方法相匹配并且最后加一个额外的参数，类型为 BlockException
* - blockHandler 函数默认需要和原方法在同一个类中
* - 若希望使用其他类的函数，则可以指定 blockHandlerClass 为对应的类的 Class 对象
* - 注意：blockHandlerClass 中对应的函数必须为 static 函数，否则无法解析
*/
protected Object handleBlockException(ProceedingJoinPoint pjp, SentinelResource annotation, BlockException ex)
throws Throwable {
// 执行 blockHandler 方法（如果配置了的话）
Method blockHandlerMethod = extractBlockHandlerMethod(pjp, annotation.blockHandler(),
annotation.blockHandlerClass());
if (blockHandlerMethod != null) {
Object[] originArgs = pjp.getArgs();
// 构造参数：原方法参数 + BlockException
Object[] args = Arrays.copyOf(originArgs, originArgs.length + 1);
args[args.length - 1] = ex;
try {
// 根据 static 方法与否进行不同的调用
if (isStatic(blockHandlerMethod)) {
return blockHandlerMethod.invoke(null, args);
}
return blockHandlerMethod.invoke(pjp.getTarget(), args);
} catch (InvocationTargetException e) {
// 抛出实际的异常
throw e.getTargetException();
}
}
// 如果没有 blockHandler，则尝试执行 fallback
return handleFallback(pjp, annotation, ex);
}
/**
* 处理 Fallback 逻辑
*
* fallback / fallbackClass 说明：
* - fallback：fallback 函数名称，可选项，用于在抛出异常的时候提供 fallback 处理逻辑
* - fallback 函数可以针对所有类型的异常（除了 exceptionsToIgnore 里面排除掉的异常类型）进行处理
*
* fallback 函数签名和位置要求：
* - 返回值类型必须与原函数返回值类型一致
* - 方法参数列表需要和原函数一致，或者可以额外多一个 Throwable 类型的参数用于接收对应的异常
* - fallback 函数默认需要和原方法在同一个类中
* - 若希望使用其他类的函数，则可以指定 fallbackClass 为对应的类的 Class 对象
* - 注意：fallbackClass 中对应的函数必须为 static 函数，否则无法解析
*/
protected Object handleFallback(ProceedingJoinPoint pjp, String fallback, String defaultFallback,
Class<?>[] fallbackClass, Throwable ex) throws Throwable {
Object[] originArgs = pjp.getArgs();
// 执行 fallback 函数（如果配置了的话）
Method fallbackMethod = extractFallbackMethod(pjp, fallback, fallbackClass);
if (fallbackMethod != null) {
// 构造参数：根据 fallback 方法的参数数量决定是否添加异常参数
int paramCount = fallbackMethod.getParameterTypes().length;
Object[] args;
if (paramCount == originArgs.length) {
args = originArgs;
} else {
args = Arrays.copyOf(originArgs, originArgs.length + 1);
args[args.length - 1] = ex;
}
try {
// 根据 static 方法与否进行不同的调用
if (isStatic(fallbackMethod)) {
return fallbackMethod.invoke(null, args);
}
return fallbackMethod.invoke(pjp.getTarget(), args);
} catch (InvocationTargetException e) {
// 抛出实际的异常
throw e.getTargetException();
}
}
// 如果没有 fallback，尝试使用 defaultFallback
return handleDefaultFallback(pjp, defaultFallback, fallbackClass, ex);
}
}
五、流控处理核心逻辑
从入口函数开始，我们深入到流控处理的核心逻辑。
入口函数调用链
public class SphU {
/**
* 创建限流入口
*
* @param name 资源名称
* @param resourceType 资源类型
* @param trafficType 流量类型（IN 或 OUT）
* @param args 参数数组
* @return Entry 对象
* @throws BlockException 如果被限流则抛出此异常
*/
public static Entry entry(String name, int resourceType, EntryType trafficType, Object[] args)
throws BlockException {
return Env.sph.entryWithType(name, resourceType, trafficType, 1, args);
}
public static Entry entry(String name, EntryType trafficType, int batchCount) throws BlockException {
return Env.sph.entry(name, trafficType, batchCount, OBJECTS0);
}
}
public class CtSph implements Sph {
@Override
public Entry entry(String name, EntryType type, int count, Object... args) throws BlockException {
StringResourceWrapper resource = new StringResourceWrapper(name, type);
return entry(resource, count, args);
}
public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
return entryWithPriority(resourceWrapper, count, false, args);
}
/**
* 带优先级的入口方法，这是限流的核心逻辑
*/
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
throws BlockException {
Context context = ContextUtil.getContext();
// 如果上下文数量超过阈值，则不进行规则检查
if (context instanceof NullContext) {
// NullContext 表示上下文数量超过了阈值，这里只初始化 Entry，不进行规则检查
return new CtEntry(resourceWrapper, null, context);
}
// 如果没有上下文，使用默认上下文
if (context == null) {
context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
}
// 如果全局开关关闭，则不进行规则检查
if (!Constants.ON) {
return new CtEntry(resourceWrapper, null, context);
}
// 获取或创建 ProcessorSlotChain（责任链）
ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
/*
* 如果资源（slot chain）数量超过 {@link Constants.MAX_SLOT_CHAIN_SIZE}，
* 则不进行规则检查
*/
if (chain == null) {
return new CtEntry(resourceWrapper, null, context);
}
// 创建 Entry 对象
Entry e = new CtEntry(resourceWrapper, chain, context);
try {
// 执行责任链进行规则检查
chain.entry(context, resourceWrapper, null, count, prioritized, args);
} catch (BlockException e1) {
// 如果被限流，释放 Entry 并抛出异常
e.exit(count, args);
throw e1;
} catch (Throwable e1) {
// 这不应该发生，除非 Sentinel 内部存在错误
log.warn("Sentinel unexpected exception,{}", e1.getMessage());
}
return e;
}
}
ProcessorSlotChain 功能插槽链
lookProcessChain 方法实际创建了 ProcessorSlotChain 功能插槽链。ProcessorSlotChain 采用责任链模式，将不同的功能（限流、降级、系统保护）组合在一起。
SlotChain 的获取与创建
ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
// 先从缓存中获取
ProcessorSlotChain chain = chainMap.get(resourceWrapper);
if (chain == null) {
// 双重检查锁，保证线程安全
synchronized (LOCK) {
chain = chainMap.get(resourceWrapper);
if (chain == null) {
// Entry 大小限制
if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
return null;
}
// 创建新的 SlotChain
chain = SlotChainProvider.newSlotChain();
// 使用不可变模式更新缓存
Map<ResourceWrapper, ProcessorSlotChain> newMap =
new HashMap<ResourceWrapper, ProcessorSlotChain>(chainMap.size() + 1);
newMap.putAll(chainMap);
newMap.put(resourceWrapper, chain);
chainMap = newMap;
}
}
}
return chain;
}
SlotChain 的构建
public class DefaultSlotChainBuilder implements SlotChainBuilder {
@Override
public ProcessorSlotChain build() {
ProcessorSlotChain chain = new DefaultProcessorSlotChain();
// 通过 SPI 加载所有 ProcessorSlot 并排序
List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);
for (ProcessorSlot slot : sortedSlotList) {
// 只处理继承自 AbstractLinkedProcessorSlot 的 Slot
if (!(slot instanceof AbstractLinkedProcessorSlot)) {
RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() +
") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
continue;
}
// 将 Slot 添加到责任链尾部
chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
}
return chain;
}
}
SlotChain 的功能划分
Slot Chain 可以分为两部分：
- 统计数据构建部分（statistic）：负责收集各种指标数据；
- 判断部分（rule checking）：根据规则判断是否限流。
官方架构图很好地解释了各个 Slot 的作用及其负责的部分。目前 ProcessorSlotChain 的设计是一个资源对应一个，构建好后缓存起来，方便下次直接取用。
各 Slot 的执行顺序
以下是 Sentinel 中各个 Slot 的默认执行顺序：
NodeSelectorSlot
↓
ClusterBuilderSlot
↓
StatisticSlot
↓
ParamFlowSlot
↓
SystemSlot
↓
AuthoritySlot
↓
FlowSlot
↓
DegradeSlot
NodeSelectorSlot - 上下文节点选择
这个功能插槽主要为资源下不同的上下文创建对应的 DefaultNode（实际用于统计指标信息）。解释一下Sentinel中的Node是什么，简单来说就是每个资源统计指标存放的容器，只不过内部由于不同的统计口径（秒级、分钟及）而分别有不同的统计窗口。Node在Sentinel不是单一的结构，而是总体上形成父子关系的树形结构。
不同的调用会有不同的 context 名称，如在当前 MVC 场景下，上下文为 sentinel_web_servlet_context。
public class NodeSelectorSlot extends AbstractLinkedProcessorSlot<Object> {
/**
* 同一个资源在不同上下文中的 DefaultNode 映射
*/
private volatile Map<String, DefaultNode> map = new HashMap<String, DefaultNode>(10);
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count,
boolean prioritized, Object... args) throws Throwable {
// 从映射表中获取当前上下文对应的节点
DefaultNode node = map.get(context.getName());
if (node == null) {
// 双重检查锁，保证线程安全
synchronized (this) {
node = map.get(context.getName());
if (node == null) {
// 创建新的 DefaultNode
node = new DefaultNode(resourceWrapper, null);
// 使用写时复制更新缓存
HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());
cacheMap.putAll(map);
cacheMap.put(context.getName(), node);
map = cacheMap;
// 构建调用树
((DefaultNode) context.getLastNode()).addChild(node);
}
}
}
// 设置当前上下文的当前节点
context.setCurNode(node);
// 继续执行后续 Slot
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
fireExit(context, resourceWrapper, count, args);
}
}
ClusterBuilderSlot - 集群节点构建
这个功能槽主要用于创建 ClusterNode。ClusterNode 和 DefaultNode 的区别是：
DefaultNode 是特定于上下文的（context-specific）；
ClusterNode 是不区分上下文的（context-independent），用于统计该资源在所有上下文中的整体数据。
public class ClusterBuilderSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
/**
* 全局 ClusterNode 映射表
*/
private static volatile Map<ResourceWrapper, ClusterNode> clusterNodeMap = new HashMap<>();
private static final Object lock = new Object();
private volatile ClusterNode clusterNode = null;
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
// 创建 ClusterNode（如果不存在）
if (clusterNode == null) {
synchronized (lock) {
if (clusterNode == null) {
// 创建集群节点
clusterNode = new ClusterNode(resourceWrapper.getName(), resourceWrapper.getResourceType());
// 更新全局映射表
HashMap<ResourceWrapper, ClusterNode> newMap =
new HashMap<>(Math.max(clusterNodeMap.size(), 16));
newMap.putAll(clusterNodeMap);
newMap.put(node.getId(), clusterNode);
clusterNodeMap = newMap;
}
}
}
// 将 ClusterNode 设置到 DefaultNode 中
node.setClusterNode(clusterNode);
// 如果有来源标识，则创建 origin node
if (!"".equals(context.getOrigin())) {
Node originNode = node.getClusterNode().getOrCreateOriginNode(context.getOrigin());
context.getCurEntry().setOriginNode(originNode);
}
// 继续执行后续 Slot
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
}
StatisticSlot - 统计插槽
StatisticSlot 是 Sentinel 最重要的类之一，用于根据规则判断结果进行相应的统计操作。
统计逻辑说明
entry 的时候：
依次执行后续的判断 Slot；
每个 Slot 触发流控会抛出异常（BlockException 的子类）；
若有 BlockException 抛出，则记录 block 数据；
若无异常抛出则算作可通过（pass），记录 pass 数据。
exit 的时候：
若无 error（无论是业务异常还是流控异常），记录 complete（success）以及 RT，线程数 -1。
记录数据的维度：
线程数 +1；
记录当前 DefaultNode 数据；
记录对应的 originNode 数据（若存在 origin）；
累计 IN 统计数据（若流量类型为 IN）。
public class StatisticSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
try {
// 此位置会调用 SlotChain 中后续的所有 Slot，完成所有规则检测
fireEntry(context, resourceWrapper, node, count, prioritized, args);
// 请求通过，增加线程数和通过数
// 代码运行到这个位置，就证明之前的所有 Slot 检测都通过了
// 此时就可以统计请求的相应数据了
// 增加线程数（+1）
node.increaseThreadNum();
// 增加通过请求的数量（这里涉及到滑动窗口算法）
node.addPassRequest(count);
// 省略其他统计逻辑...
} catch (PriorityWaitException ex) {
// 如果是优先级等待异常，记录优先级等待数
node.increaseThreadNum();
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseThreadNum();
}
if (resourceWrapper.getEntryType() == EntryType.IN) {
// 记录入站统计数据
Constants.ENTRY_NODE.increaseThreadNum();
}
throw ex;
} catch (BlockException e) {
// 如果被限流，记录被限流数
// 省略 block 统计逻辑...
throw e;
} catch (Throwable ex) {
// 如果发生业务异常，记录异常数
// 省略异常统计逻辑...
throw ex;
}
}
@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
// 若无 error（无论是业务异常还是流控异常），记录 complete（success）以及 RT，线程数-1
// 记录数据的维度：线程数+1、记录当前 DefaultNode 数据、记录对应的 originNode 数据（若存在 origin）
// 、累计 IN 统计数据（若流量类型为 IN）
// 省略 exit 统计逻辑...
}
}
StatisticNode 数据结构
到这里，StatisticSlot 的作用已经比较清晰了。接下来我们需要分析它的统计数据结构。fireEntry 调用向下的节点和之前的方式一样，剩下的节点主要包括：
- ParamFlowSlot；
- SystemSlot；
- AuthoritySlot；
- FlowSlot；
- DegradeSlot；
其中比较常见的是流控和熔断：FlowSlot、DegradeSlot，所以下面我们着重分析 FlowSlot。
六、FlowSlot - 流控插槽
这个 Slot 主要根据预设的资源的统计信息，按照固定的次序依次生效。如果一个资源对应两条或者多条流控规则，则会根据如下次序依次检验，直到全部通过或者有一个规则生效为止。
FlowSlot 核心逻辑
@SpiOrder(-2000)
public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
// 执行流控检查
checkFlow(resourceWrapper, context, node, count, prioritized);
// 继续执行后续 Slot
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
// 省略其他方法...
}
checkFlow 方法详解
/**
* 执行流控检查
*
* @param ruleProvider 规则提供者函数
* @param resource 资源包装器
* @param context 上下文
* @param node 节点
* @param count 请求数量
* @param prioritized 是否优先
* @throws BlockException 如果被限流则抛出异常
*/
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 继承 BlockException
throw new FlowException(rule.getLimitApp(), rule);
}
}
}
}
通过这里我们就可以得知，流控规则是通过 FlowRule 来完成的，数据来源是我们使用的流控控制台，也可以通过代码进行设置。
FlowRule 流控规则
每条流控规则主要由三个要素构成：
- grade（阈值类型）：按 QPS（每秒请求数）还是线程数进行限流;
- strategy（调用关系策略）：基于调用关系的流控策略;
- controlBehavior（流控效果）：当 QPS 超过阈值时的流量整形行为。
public class FlowRule extends AbstractRule {
public FlowRule() {
super();
// 来源默认 Default
setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
}
public FlowRule(String resourceName) {
super();
// 资源名称
setResource(resourceName);
setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
}
/**
* 流控的阈值类型
* 0: 线程数
* 1: QPS
*/
private int grade = RuleConstant.FLOW_GRADE_QPS;
/**
* 流控阈值
*/
private double count;
/**
* 基于调用链的流控策略
* STRATEGY_DIRECT: 直接流控（按来源）
* STRATEGY_RELATE: 关联流控（关联资源）
* STRATEGY_CHAIN: 链路流控（按入口资源）
*/
private int strategy = RuleConstant.STRATEGY_DIRECT;
/**
* 关联流控模式下的关联资源
*/
private String refResource;
/**
* 流控效果（流量整形行为）
* 0: 默认（直接拒绝）
* 1: 预热（Warm Up）
* 2: 排队等待（Rate Limiter）
* 3: 预热 + 排队等待（目前控制台没有）
*/
private int controlBehavior = RuleConstant.CONTROL_BEHAVIOR_DEFAULT;
/**
* 预热时长（秒）
*/
private int warmUpPeriodSec = 10;
/**
* 排队等待的最大超时时间（毫秒）
*/
private int maxQueueingTimeMs = 500;
/**
* 是否为集群模式
*/
private boolean clusterMode;
/**
* 集群模式配置
*/
private ClusterFlowConfig clusterConfig;
/**
* 流量整形控制器
*/
private TrafficShapingController controller;
// 省略 getter/setter 方法...
}
七、滑动窗口算法
不管流控规则采用何种流控算法，在底层都需要有支持指标统计的数据结构作为支撑。在 Sentinel 中，用于支撑基于 QPS 等限流的数据结构是 StatisticNode。
StatisticNode 数据结构
public class StatisticNode implements Node {
/**
* 保存最近 1 秒内的统计数据
* 每个桶（bucket）500ms，共 2 个桶
*/
private transient volatile Metric rollingCounterInSecond =
new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);
/**
* 保存最近 60 秒的统计数据
* windowLengthInMs 被特意设置为 1000 毫秒，即每个桶代表 1 秒
* 共 60 个桶，这样可以获得每秒精确的统计信息
*/
private transient Metric rollingCounterInMinute =
new ArrayMetric(60, 60 * 1000, false);
// 省略其他字段和方法...
}
ArrayMetric 核心实现
ArrayMetric 是 Sentinel 中数据采集的核心，内部使用了 BucketLeapArray，即滑动窗口的思想进行数据的采集。
public class ArrayMetric implements Metric {
/**
* 滑动窗口数组
*/
private final LeapArray<MetricBucket> 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);
}
}
}
这里有两种实现：
-
BucketLeapArray：普通滑动窗口，每个时间桶仅记录固定时间窗口内的指标数据；
-
OccupiableBucketLeapArray：扩展实现，支持"抢占"未来时间窗口的令牌或容量，在流量突发时允许借用后续窗口的配额，实现更平滑的限流效果。
BucketLeapArray - 滑动窗口实现
LeapArray 核心属性
LeapArray 是滑动窗口的基础类，其核心属性如下：
/**
* 窗口大小（长度），单位：毫秒
* 例如：1000ms
*/
private int windowLengthInMs;
/**
* 样本数（桶的数量）
* 例如：5（表示 5 个桶，每个 1000ms，总共 5 秒）
*/
private int sampleCount;
/**
* 采集周期（总时间窗口长度），单位：毫秒
* 例如：5 * 1000ms（5 秒）
*/
private int intervalInMs;
/**
* 窗口数组，array 长度就是样本数 sampleCount
*/
protected final AtomicReferenceArray<WindowWrap<T>> array;
/**
* 更新窗口数据的锁，保证数据的正确性
*/
private final ReentrantLock updateLock;
WindowWrap 窗口包装器
每个窗口包装器包含三个属性：
public class WindowWrap<T> {
/**
* 窗口大小（长度），单位：毫秒
* 与 LeapArray 中的 windowLengthInMs 一致
*/
private final long windowLengthInMs;
/**
* 窗口开始时间戳
* 它的值是 windowLengthInMs 的整数倍
*/
private long windowStart;
/**
* 窗口数据（泛型 T）
* Sentinel 目前只有 MetricBucket 类型，存储统计数据
*/
private T value;
}
MetricBucket 指标桶
public class MetricBucket {
/**
* 计数器数组
* 长度是需要统计的事件种类数，目前是 6 个
* LongAdder 是线程安全的计数器，性能优于 AtomicLong
*/
private final LongAdder[] counters;
// 省略其他字段和方法...
}
滑动窗口工作原理
LeapArray 统计数据的基本思路：
创建一个长度为 n 的数组，数组元素就是窗口；
每个窗口包装了 1 个指标桶，桶中存放了该窗口时间范围内对应的请求统计数据；
可以想象成一个环形数组在时间轴上向右滚动；
请求到达时，会命中数组中的一个窗口，该请求的数据就会存到命中的这个窗口包含的指标桶中；
当数组转满一圈时，会回到数组的开头；
此时下标为 0 的元素需要重复使用，它里面的窗口数据过期了，需要重置，然后再使用。
获取当前窗口
LeapArray 获取当前时间窗口的方法：
/**
* 获取当前时间戳对应的窗口
*
* @return 当前时间的窗口
*/
public WindowWrap<T> currentWindow() {
return currentWindow(TimeUtil.currentTimeMillis());
}
/**
* 获取指定时间戳对应的窗口（核心方法）
*
* @param timeMillis 时间戳（毫秒）
* @return 对应的窗口
*/
public WindowWrap<T> currentWindow(long timeMillis) {
if (timeMillis < 0) {
return null;
}
// 计算数组下标
int idx = calculateTimeIdx(timeMillis);
// 计算当前请求对应的窗口开始时间
long windowStart = calculateWindowStart(timeMillis);
// 无限循环，确保能够获取到窗口
while (true) {
// 取窗口
WindowWrap<T> old = array.get(idx);
if (old == null) {
// 第一次使用，创建新窗口
WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
// CAS 操作，确保只初始化一次
if (array.compareAndSet(idx, null, window)) {
// 成功更新，返回创建的窗口
return window;
} else {
// CAS 失败，让出时间片，等待其他线程完成初始化
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()) {
// 异常情况：机器时钟回拨等
// 正常情况不会进入该分支
return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
}
}
}
数据存储
在获取到窗口之后，就可以存储数据了。ArrayMetric 实现了 Metric 中存取数据的接口方法。
示例：存储 RT（响应时间）
/**
* 添加响应时间数据
*
* @param rt 响应时间（毫秒）
*/
public void addRT(long rt) {
// 获取当前时间窗口，data 为 BucketLeapArray
WindowWrap<MetricBucket> wrap = data.currentWindow();
// 计数
wrap.value().addRT(rt);
}
/**
* MetricBucket 的 addRT 方法
*
* @param rt 响应时间
*/
public void addRT(long rt) {
// 记录 RT 时间对 rt 值
add(MetricEvent.RT, rt);
// 记录最小响应时间（非线程安全，但没关系）
if (rt < minRt) {
minRt = rt;
}
}
/**
* 通用的计数方法
*
* @param event 事件类型
* @param n 增加的数量
* @return 当前桶
*/
public MetricBucket add(MetricEvent event, long n) {
counters[event.ordinal()].add(n);
return this;
}
数据读取
示例：读取 RT（响应时间）
/**
* 获取总响应时间
*
* @return 总响应时间
*/
public long rt() {
// 触发当前窗口更新（处理过期窗口）
data.currentWindow();
long rt = 0;
// 取出所有的 bucket
List<MetricBucket> list = data.values();
for (MetricBucket window : list) {
rt += window.rt(); // 求和
}
return rt;
}
/**
* 获取所有有效的窗口
*
* @return 有效窗口列表
*/
public List<T> values() {
return values(TimeUtil.currentTimeMillis());
}
/**
* 获取指定时间之前的所有有效窗口
*
* @param timeMillis 时间戳
* @return 有效窗口列表
*/
public List<T> values(long timeMillis) {
if (timeMillis < 0) {
return new ArrayList<T>(); // 正常情况不会到这里
}
int size = array.length();
List<T> result = new ArrayList<T>(size);
for (int i = 0; i < size; i++) {
WindowWrap<T> windowWrap = array.get(i);
// 过滤掉没有初始化过的窗口和过期的窗口
if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
continue;
}
result.add(windowWrap.value());
}
return result;
}
/**
* 判断窗口是否过期
*
* @param time 给定时间（通常是当前时间）
* @param windowWrap 窗口包装器
* @return 如果过期返回 true
*/
public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {
// 给定时间与窗口开始时间超过了一个采集周期
return time - windowWrap.windowStart() > intervalInMs;
}
OccupiableBucketLeapArray - 可抢占窗口
为什么需要 OccupiableBucketLeapArray？
假设一个资源的访问 QPS 稳定是 10，请求是均匀分布的：
在时间 0.0-1.0 秒区间中，通过了 10 个请求；
在 1.1 秒的时候，观察到的 QPS 可能只有 5，因为此时第一个时间窗口被重置了，只有第二个时间窗口有值；
当在秒级统计的情形下，用 BucketLeapArray 会有 0~50%的数据误这时就要用 OccupiableBucketLeapArray 来解决这个问题。
OccupiableBucketLeapArray 实现
从上面我们可以看到在秒级统计 rollingCounterInSecond 中，初始化实例时有两种构造参数：
public class OccupiableBucketLeapArray extends LeapArray<MetricBucket> {
/**
* 借用未来窗口的数组
*/
private final FutureBucketLeapArray borrowArray;
public OccupiableBucketLeapArray(int sampleCount, int intervalInMs) {
super(sampleCount, intervalInMs);
// 创建借用窗口数组
this.borrowArray = new FutureBucketLeapArray(sampleCount, intervalInMs);
}
/**
* 创建新的空桶
* 会从 borrowArray 中借用数据
*/
@Override
public MetricBucket newEmptyBucket(long time) {
MetricBucket newBucket = new MetricBucket();
// 获取借用窗口的数据
MetricBucket borrowBucket = borrowArray.getWindowValue(time);
if (borrowBucket != null) {
// 将借用数据复制到新桶中
newBucket.reset(borrowBucket);
}
return newBucket;
}
/**
* 重置窗口
* 会从 borrowArray 中借用 pass 数据
*/
@Override
protected WindowWrap<MetricBucket> resetWindowTo(WindowWrap<MetricBucket> w, long time) {
// 更新开始时间并重置值
w.resetTo(time);
MetricBucket borrowBucket = borrowArray.getWindowValue(time);
if (borrowBucket != null) {
// 重置桶值并添加借用的 pass 数据
w.value().reset();
w.value().addPass((int) borrowBucket.pass());
} else {
w.value().reset();
}
return w;
}
/**
* 获取当前等待中的请求数量
*/
@Override
public long currentWaiting() {
borrowArray.currentWindow();
long currentWaiting = 0;
List<MetricBucket> list = borrowArray.values();
for (MetricBucket window : list) {
currentWaiting += window.pass();
}
return currentWaiting;
}
/**
* 添加等待中的请求数量
*
* @param time 时间
* @param acquireCount 获取数量
*/
@Override
public void addWaiting(long time, int acquireCount) {
WindowWrap<MetricBucket> window = borrowArray.currentWindow(time);
window.value().add(MetricEvent.PASS, acquireCount);
}
}
八、总结
至此，Sentinel 的基本情况都已经分析完成。以上内容主要讲解了 Sentinel 的核心处理流程，包括：
核心流程总结
- 规则加载：
- 通过
RuleLoader 将各种规则（流控、熔断、系统限流等）加载到内存缓存中。
- 请求拦截：
- 通过 SentinelServletFilter 过滤器拦截 HTTP 请求；
- 通过SentinelResourceAspect切面处理 @SentinelResource 注解。
- 责任链处理：
- 使用 ProcessorSlotChain 责任链模式组合多个功能插槽；
- 每个插槽负责特定的功能（统计、流控、熔断等）。
- 流控判断：
- FlowSlot 根据流控规则判断是否限流；
- 通过滑动窗口算法统计 QPS、线程数等指标。
- 异常处理：
- 被限流时抛出 BlockException；
- 通过 blockHandler 或 fallback 处理异常。
核心技术点
- 责任链模式：
- 通过
ProcessorSlotChain 将不同的限流功能组合在一起。
- 滑动窗口算法：
- LeapArray 实现环形滑动窗口；
- BucketLeapArray 普通滑动窗口；
- OccupiableBucketLeapArray 可抢占窗口，支持借用未来配额。
- 数据结构：
- DefaultNode：特定于上下文的统计节点；
- ClusterNode：不区分上下文的集群统计节点；
- StatisticNode：核心统计节点，包含秒级和分钟级统计。
- 限流算法：
- QPS 限流：通过滑动窗口统计 QPS；
- 线程数限流：通过原子计数器统计线程数；
- 流控效果：快速失败、预热、排队等待等；
Sentinel 通过精心设计的架构，实现了高效、灵活、可扩展的流量控制能力，为微服务系统提供了强大的保护机制。
往期回顾
1.社区推荐重排技术：双阶段框架的实践与演进｜得物技术
2.Flink ClickHouse Sink：生产级高可用写入方案｜得物技术
3.服务拆分之旅：测试过程全揭秘｜得物技术
4.大模型网关：大模型时代的智能交通枢纽｜得物技术
5.从“人治”到“机治”：得物离线数仓发布流水线质量门禁实践
文 /万钧
关注得物技术，每周更新技术干货
要是觉得文章对你有帮助的话，欢迎评论转发点赞～
未经得物技术许可严禁转载，否则依法追究法律责任。
登录