1、前言

在nacos2.3.0中，新增了反脆弱功能插件。引用官网的话：反脆弱是对访问服务端的某种资源频率和次数达到一定程度时进行的限制访问的策略，用于保护服务端在高压情况下能快速拒绝请求，防止过多的资源访问导致服务端资源耗尽引起的大面积不可用；Nacos反脆弱插件，将信息主要抽象为监控点和反脆弱规则。 (还是第一次听说到反脆弱这个名词)

地址链接：nacos.io/docs/v2/plu…

2、nacos的启用限流注解@TpsControl

@Retention(RetentionPolicy.RUNTIME)
public @interface TpsControl {
    /**
     * alias name for control point.
     * 监控点别名
     * @return
     */
    String name() default "";
    
    /**
     * The point name should applied for.
     * 监控点名称
     * @return action type, default READ
     */
    String pointName();
    
}

@TpsControl(pointName = "NamingInstanceRegister", name = "HttpNamingInstanceRegister")
@TpsControl(pointName = "NamingServiceUpdate", name = "HttpNamingServiceUpdate")

@TpsControl注解有两个属性，name和pointName，基本可以认为就一个，即监控点名称。监控点详情可查看官网介绍。

反脆弱拦截入口：NacosHttpTpsFilter，通过servlet的过滤器进行过滤拦截。

@Override
public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain)
        throws IOException, ServletException {
    final HttpServletRequest httpServletRequest = (HttpServletRequest) servletRequest;
    final HttpServletResponse response = (HttpServletResponse) servletResponse;
    
    Method method = controllerMethodsCache.getMethod(httpServletRequest);
    try {
        if (method != null && method.isAnnotationPresent(TpsControl.class)
                && TpsControlConfig.isTpsControlEnabled()) {
            TpsControl tpsControl = method.getAnnotation(TpsControl.class);
            String pointName = tpsControl.pointName();
            String parserName = StringUtils.isBlank(tpsControl.name()) ? pointName : tpsControl.name();
            HttpTpsCheckRequestParser parser = HttpTpsCheckRequestParserRegistry.getParser(parserName);
            TpsCheckRequest httpTpsCheckRequest = null;
            if (parser != null) {
                httpTpsCheckRequest = parser.parse(httpServletRequest);
            }
            if (httpTpsCheckRequest == null) {
                httpTpsCheckRequest = new TpsCheckRequest();
            }
            if (StringUtils.isBlank(httpTpsCheckRequest.getPointName())) {
                httpTpsCheckRequest.setPointName(pointName);
            }
            initTpsControlManager();
            TpsCheckResponse checkResponse = tpsControlManager.check(httpTpsCheckRequest);
            if (!checkResponse.isSuccess()) {
                AsyncContext asyncContext = httpServletRequest.startAsync();
                asyncContext.setTimeout(0);
                RpcScheduledExecutor.CONTROL_SCHEDULER.schedule(
                        () -> generate503Response(httpServletRequest, response, checkResponse.getMessage(),
                                asyncContext), 1000L, TimeUnit.MILLISECONDS);
                return;
            }
            
        }
    } catch (Throwable throwable) {
        Loggers.TPS.warn("Fail to  http tps check", throwable);
    }
    
    filterChain.doFilter(httpServletRequest, response);
}

整个doFilter方法也比较简单。

1. 先是根据请求获取处理该请求controller对应的方法。
2. 判断方法上是否标注了TpsControl注解；只处理标注了该注解的方法。
3. 封装实体。
4. 判断校验是否进行反脆弱过滤。

3、nacos固定窗口限流的设计

Nacos默认使用简单的同秒统计方式，即按照时钟的秒来进行统计。

其中核心实现类为LocalSimpleCountRateCounter。

public LocalSimpleCountRateCounter(String name, TimeUnit period) {
    super(name, period);
    // 初始化插槽list，默认是10
    slotList = new ArrayList<>(DEFAULT_RECORD_SIZE);
    for (int i = 0; i < DEFAULT_RECORD_SIZE; i++) {
        slotList.add(new TpsSlot());
    }
    long now = System.currentTimeMillis();
    // 初始化开始时间startTime，忽略毫秒
    if (period == TimeUnit.SECONDS) {
        startTime = RateCounter.getTrimMillsOfSecond(now);
    } else if (period == TimeUnit.MINUTES) {
        startTime = RateCounter.getTrimMillsOfMinute(now);
    } else if (period == TimeUnit.HOURS) {
        startTime = RateCounter.getTrimMillsOfHour(now);
    } else {
        //second default
        startTime = RateCounter.getTrimMillsOfSecond(now);
    }
}

LocalSimpleCountRateCounter继承自RateCounter，RateCounter是抽象类，主要定义了一些必要的方法。其中核心方法tryAdd，是否要对该请求进行限流；

先看两个重要的静态内部类TpsSlot和SlotCountHolder

static class TpsSlot {
    // 第一个请求到达的时间
    long time = 0L;
    // 插槽计数器
    private SlotCountHolder countHolder = new SlotCountHolder();
    // 重置 把第一个请求到达的时间更新为传入的时间；重置插槽计数器的值
    public void reset(long second) {
        synchronized (this) {
            if (this.time != second) {
                this.time = second;
                countHolder.count.set(0L);
                countHolder.interceptedCount.set(0);
            }
        }
    }
    
    @Override
    public String toString() {
        return "TpsSlot{" + "time=" + time + ", countHolder=" + countHolder + '}';
    }
    
}

// 插槽计数器
static class SlotCountHolder {
    // 当前已经请求的数据（已放行的请求）；AtomicLong 保证并发场景下的原子性和可见性
    AtomicLong count = new AtomicLong();
    // 当前被限流拦截的请求数据
    AtomicLong interceptedCount = new AtomicLong();
    
    @Override
    public String toString() {
        return "{" + count + "|" + interceptedCount + '}';
    }
}

 /**
   * @param timestamp: 请求调用时的时间戳
   * @param countDelta: 请求数，默认1
   * @param upperLimit: 最大请求数，例如果配置每秒最大请求数100，则upperLimit为100
   * @return boolean true:请求数不超过最大请求数，false:请求数超过最大请求数
   */
  @Override
  public boolean tryAdd(long timestamp, long countDelta, long upperLimit) {
      // 根据请求的时间戳获取对应的插槽，CAS尝试增加请求数
      if (createSlotIfAbsent(timestamp).countHolder.count.addAndGet(countDelta) <= upperLimit) {
          return true;
      } else {
          // 如果请求数超过最大请求数，则记录被拒绝的请求数
          createSlotIfAbsent(timestamp).countHolder.interceptedCount.addAndGet(countDelta);
          return false;
      }
  }

public TpsSlot createSlotIfAbsent(long timeStamp) {
    // 当前请求时间-开始时间 示例：startTime为1716901705000，当前请求时间戳为1716901708911，则distance为3911，即相差3911毫秒
    long distance = timeStamp - startTime;
    // 这里以秒为例，忽略毫秒，计算相差的秒数diff=3
    long diff = (distance < 0 ? distance + getPeriod().toMillis(1) * DEFAULT_RECORD_SIZE : distance) / getPeriod()
            .toMillis(1);
    // 计算当前的窗口时间 currentWindowTime = 1716901705000+3*1000=1716901708000
    long currentWindowTime = startTime + diff * getPeriod().toMillis(1);
    // 根据diff%10计算出对应的插槽index，此时index=3
    int index = (int) diff % DEFAULT_RECORD_SIZE;
    // 取出对应的插槽
    TpsSlot tpsSlot = slotList.get(index);
    // 判断当前插槽对应的时间是否为当前窗口对应的时间，如果不为当前窗口，则重置插槽对应的时间为当前窗口对应的时间
    if (tpsSlot.time != currentWindowTime) {
        tpsSlot.reset(currentWindowTime);
    }
    return slotList.get(index);
}

示例流程：

1. 假设初始化时间startTime=1716901705000（限流设计中使用到的时间戳全部忽略毫秒）
2. 若第一次请求timeStamp=1716901705435，计算与startTime相差时间为0秒，用0对插槽总数量10取余为0，则这次请求的计数器落在0号插槽上进行计数，此时插槽未初始化，初始化插槽time=1716901705000，count=0，interceptedCount=0，对0号插槽count进行CAS加1（1716901705000到1716901706000这1秒钟所有的请求都会由0号插槽进行计数）
3. 若第二次请求timeStamp=1716901725333，计算与startTime相差时间为20秒，用20对10取余为0，则这次请求的计数器落在0号插槽上进行计数，此时0号插槽time=1716901705000，不等于1716901725000（忽略毫秒后），则重置0号插槽time=1716901725000，count=0，interceptedCount=0，对0号插槽count进行CAS加1（1716901725000到1716901726000这1秒钟所有的请求都会由0号插槽进行计数）

设计亮点：

• 不依赖于任何第三方组件，完全基于本地内存。
• CAS乐观锁，提高效率；但每秒的请求数量如果特别大，CPU损耗过高。
• 分而治之，插槽的设计让连续的10s内每秒请求的计数肯定落在不同的插槽上，提高性能。避免了所有请求对一个资源CAS，类似个人感觉利大于弊。

4、最后

正如官网所介绍的：对于大多数场景来说是足够使用的，但对于一些精确度要求高的用户而言，可能需要使用滑动窗口等更精确的方式进行统计。固定窗口的限流算法也是相当巧妙。