常见的四种限流算法

近期处理了关于限流相关的问题,了解到有四种限流的算法,以下的代码使用的是Redis缓存,将此次的使用进行一个记录,以下是相关的代码:

限流的方式

1.固定窗口算法(计数器)

介绍:

• 固定窗口又称固定窗口（又称计数器算法，Fixed Window）限流算法，是最简单的限流算法，通过在单位时间内维护的计数器来控制该时间单位内的最大访问量。

优点:

• 固定窗口算法非常简单，易于实现和理解

缺点:

• 存在明显的临界问题,当两个窗口交界处，瞬时流量可能为2n,如图所示:限流阀值为2个请求,单位时间窗口是1s,在单位时间内的前0.7-1s和1-1.4s，
• 分别并发2个请求,虽然在时间窗口期间都没有超过阀值，但是如果算0.7-1.4s,则并发数为4个，已经超过单位时间1s不超过2阀值的定义

原理图:

产生的问题:

固定窗口算法代码:

/**
 * 固定窗口
 */
public boolean windowRate(RateLimiter rateLimiter, RateLimiterData rateLimiterData,String combineKey){

    // 获取当前时间
    long currentTime = System.currentTimeMillis();
    // 判断是否在窗口期
    if (currentTime - rateLimiterData.getStartTime() > TimeUnit.SECONDS.toMillis(rateLimiter.time())) {
        //重置计数器
        rateLimiterData.setRequestsMax(0);
        rateLimiterData.setStartTime(currentTime);
        log.warn("startTime:"+rateLimiterData.getStartTime());
    }

    // 最大请求数量大于计数器数量
    if (rateLimiter.count() > rateLimiterData.getRequestsMax() ) {
        // 计数器自增
        Integer count=rateLimiterData.getRequestsMax();

        rateLimiterData.setRequestsMax(count+1);
        log.warn("requestsMax:"+rateLimiterData.getRequestsMax());
        redisCache.setCacheObject(combineKey, rateLimiterData,20,TimeUnit.SECONDS);

        return true;

    }
    return false;

}

2.滑动窗口算法

介绍:

• 为了防止瞬时流量，可以把固定窗口近一步划分成多个格子，每次向后移动一小格，而不是固定窗口大小，可以解决固定窗口临界值的问题,但是无法避免。这就是滑动窗口（Sliding Window）。

优点:

• 可以消除计数器算法跨周期时的流量高峰。周期内的间隔划分越多，流量越平稳。

缺点:

• 无法缓存多余的请求,需要合理调整时间窗口大小,流量的数量达到阈值时会瞬间掐断流量，会导致流量不够平滑.

原理图:

第一次的请求发起时间为:14:30:00 窗口时间为:14:30:00-14:30:02
第二次的请求发起时间为:14:30:01 窗口时间为:14:30:00-14:30:02
第三次的请求发起时间为:14:30:03 窗口时间为:14:30:02-14:30:04
第四次的请求发起时间为:14:30:05 窗口时间为:14:30:04-14:30:06

滑动窗口相比较固定窗口是如何减少临界值的影响:
从图上可以看到时间创建是一种向右滑动的方式前进, 滑动窗口限流能够显著减少临界问题产生的影响,但是不能消除掉。
窗口每次向前滑动一个单元格,当请求到达时，只要窗口中所有单元格的计数总和不超过阈值都可以放行, 假如2s秒限流4个 那么就是也就是滑动窗口（单位时间）被划分为4个小周期。每格表示0.5S。每过0.5s时间窗口就会往右滑动一格。而且每个小周期，都有自己独立的计数器
具体的示例以及固定窗口和滑动窗口的区别:
14:30:00-14:30:02
滑动窗口一:0-0.5
滑动窗口二0.5-1
滑动窗口三:1-1.5
滑动窗口四:1.5-2
此时分别在1.8s来了5个请求,滑动窗口三已经限流,2.6s又来了5个请求,如果是固定算法的话会不限流,
但是过了1s会向右移动两个周期,当前的时间窗口单位2.5-3s,2.6处于这个时间段之内,这个区域的数量已经超过限定.就会触发限流。
滑动算法的周期划分越多,那么滚动的就会越平滑,限流的统计数量则越准确;

问题:
但是临界值的问题无法消除,只是缩小了范围而已;

滑动窗口算法代码:

public boolean  slidingWindowRateState(RateLimiter rateLimiter,String combineKey){
       TreeMap<Long, Integer> counters = new TreeMap<>();
       counters=redisCache.getCacheObject(combineKey);
       if(counters==null){
           counters=new TreeMap<>();
       }
       boolean state=slidingWindowRate(rateLimiter,counters,combineKey);

       return state;
   }
   
   /**
    * 滑动窗口
    */
   public boolean  slidingWindowRate(RateLimiter rateLimiter, TreeMap<Long, Integer> counters,String combineKey){
       // 获取当前时间
       long currentTime = System.currentTimeMillis();
       //窗口的开始时间戳
       long slideTime=currentTime-TimeUnit.SECONDS.toMillis(rateLimiter.time());
       //当前窗口总请求数
       int requestsCount = 0;
       //遍历存储的计数器
       Iterator<Map.Entry<Long, Integer>> iterator = counters.entrySet().iterator();
       while (iterator.hasNext()) {
           Map.Entry<Long, Integer> entry = iterator.next();
           // 删除过期的子窗口计数器
           if (entry.getKey() < slideTime) {
               iterator.remove();
           } else {
               //当前窗口时间的所有计数器进行累加
               requestsCount = requestsCount + entry.getValue();
           }
       }

       if (requestsCount >= rateLimiter.count()) {
           return false;
       }else {
           //计数器增加1
           counters.merge(currentTime, 1, Integer::sum);
           redisCache.setCacheObject(combineKey, counters,20,TimeUnit.SECONDS);
           return true;
       }


   }

3.漏桶算法

介绍:

• 漏桶限流算法（Leaky Bucket Algorithm）是一种流量控制算法，用于控制流入网络的数据速率，以防止网络拥塞。它的思想是将数据包看作是水滴，漏桶看作是一个固定容量的水桶，数据包像水滴一样从桶的顶部流入桶中，并通过桶底的一个小孔以一定的速度流出，从而限制了数据包的流量。

优点:

• 令牌桶算法是一种流量控制机制，非常适合于处理突发流量，同时保证一定程度的平滑流动。它的工作原理类似于一个实际的水桶，其中水桶代表令牌桶，水流代表令牌。令牌以恒定的速率填充到桶中，直到达到桶的容量上限，多余的> 令牌会被丢弃。
• 可以平滑限制请求的处理速度，避免瞬间请求过多导致系统崩溃或者雪崩。
• 可以控制请求的处理速度，使得系统可以适应不同的流量需求，避免过载或者过度闲置。
• 可以通过调整桶的大小和漏出速率来满足不同的限流需求，可以灵活地适应不同的场景。

缺点:

• 短时间内有突发的大流量时，漏桶可能处理不了这些请求,还有就是同时放行多个请求的时候会导致业务处理压力比较大。

原理图:

问题:
因为速率是固定的无法处理瞬时利流量,不能很快处理请求

漏桶算法代码:

/**
   * 漏桶
   */
  public boolean leakyBucket(RateLimiter rateLimiter, RateLimiterData rateLimiterData,String combineKey){
      //漏水速率，单位:个/秒
      int rate = (int) Math.ceil( rateLimiter.count() / rateLimiter.time());
      rateLimiterData=redisCache.getCacheObject(combineKey);
      long requestsTime = System.currentTimeMillis();
          // 时间间隔=本次请求的时间和上次请求的时间
          if(rateLimiterData.getStartTime()==0){
              rateLimiterData.setStartTime(requestsTime);
          }
          long leakDuration = requestsTime - rateLimiterData.getStartTime();
              // 计算漏出的水量
              long leakWater = leakDuration / TimeUnit.SECONDS.toMillis(1) * rate;
              if(leakWater>0){
                  // 漏桶漏出水后,更新桶里面的数量
                  int waterNum = (int)Math.max(0, rateLimiterData.getRequestsMax() - leakWater);
                  rateLimiterData.setRequestsMax(waterNum);
                  rateLimiterData.setStartTime(requestsTime);
              }

          // 判断桶中的水数量大于请求的数量,大于则添加水
          if (rateLimiter.count()>rateLimiterData.getRequestsMax()) {
              rateLimiterData.setRequestsMax(rateLimiterData.getRequestsMax()+rateLimiter.deduction());
              redisCache.setCacheObject(combineKey, rateLimiterData,20,TimeUnit.SECONDS);
              return true;
          }
           return false;

  }

4.令牌桶算法

介绍:

• 令牌桶算法是一种常用的限流算法，可以用于限制单位时间内请求的数量。该算法维护一个固定容量的令牌桶，每秒钟会向令牌桶中放入一定数量的令牌。当有请求到来时，如果令牌桶中有足够的令牌，则请求被允许通过并从令牌桶中消耗一个令牌，否则请求被拒绝。

优点:

• 稳定性高：令牌桶算法可以控制请求的处理速度，可以使系统的负载变得稳定。
• 精度高：令牌桶算法可以根据实际情况动态调整生成令牌的速率，可以实现较高精度的限流。
• 弹性好：令牌桶算法可以处理突发流量，可以在短时间内提供更多的处理能力，以处理突发流量。

缺点:

• 实现复杂：令牌桶算法的实现较为复杂。对短时请求难以处理：在短时间内有大量请求到来时，可能会导致令牌桶中的令牌被快速消耗完，从而限流。这种情况下，可以考虑使用漏桶算法。
• 时间精度要求高：令牌桶算法要求时间精度较高，因为是根据时间来生成令牌的数量,如果系统时间出现错误，会导致限流效果不理想。

原理图:

令牌桶算法代码:

/**
    * 令牌桶
    * @param rateLimiter
    * @param rateLimiterData
    * @param combineKey
    * @return
    */
   public boolean tokenBucket(RateLimiter rateLimiter, RateLimiterData rateLimiterData,String combineKey){
       //令牌生成的速率，单位:个/秒
       int rate = (int) Math.ceil( rateLimiter.count() / rateLimiter.time());
       rateLimiterData=redisCache.getCacheObject(combineKey);
       //请求时间
       long requestsTime = System.currentTimeMillis();
           if(rateLimiterData.getStartTime()==0){
               rateLimiterData.setRequestsMax(rateLimiter.count()-rateLimiter.deduction());
               rateLimiterData.setStartTime(requestsTime);
               redisCache.setCacheObject(combineKey, rateLimiterData,20,TimeUnit.SECONDS);
           }else {
               // 时间间隔=本次请求的时间和上次请求的时间
               long residueTime = requestsTime - rateLimiterData.getStartTime();
               // 计算令牌产生的数量
               int tokenNum = (int) (residueTime/ TimeUnit.SECONDS.toMillis(1) * rate);
               //总数量=剩余的数量+生成的数量
               Integer generateToken= Math.abs(rateLimiterData.getRequestsMax() + tokenNum);
               //最后剩余的数量
               int residueNum = Math.min(rateLimiterData.getRequestsMax(), generateToken);
               //剩余数量
               rateLimiterData.setRequestsMax(residueNum);
               rateLimiterData.setStartTime(requestsTime);
               //判断剩余的令牌数量是否大于等于扣减的数量
               if (rateLimiterData.getRequestsMax() >= rateLimiter.deduction()) {
                   rateLimiterData.setRequestsMax(rateLimiterData.getRequestsMax()-rateLimiter.deduction());
                   redisCache.setCacheObject(combineKey, rateLimiterData,20,TimeUnit.SECONDS);
                   return true;
               }
           }

       return false;
   }

公共的参数

RateLimiterAspect.java

/**
 * 限流
 */
@Aspect
@Component
@Slf4j
public class RateLimiterAspect {


    @Autowired
    private RedisCache redisCache;

    @Before("@annotation(rateLimiter)")
    public void doBefore(JoinPoint point, RateLimiter rateLimiter) throws Throwable {
        //厂商信息
        LimitResultType limitResultType = rateLimiter.limitResultType();
        //返回值类型
        String resultType = rateLimiter.resultType();
        String luaScript = buildScript();

        //获得key
        String combineKey = getCombineKey(rateLimiter, point);
        boolean state = false;
        /*****************开始*****************/
        RateLimiterData rateLimiterData = new RateLimiterData();
        rateLimiterData = redisCache.getCacheObject(combineKey);
        if (rateLimiterData == null) {
            RateLimiterData rateLimiterData1 = new RateLimiterData();
            rateLimiterData1.setRequestsMax(0);
            rateLimiterData1.setStartTime(0);
            rateLimiterData = rateLimiterData1;
        }
        //固定窗口
        state = windowRate(rateLimiter, rateLimiterData, combineKey);
        //漏桶
        state = leakyBucket(rateLimiter, rateLimiterData, combineKey);
        //令牌
        state = tokenBucket(rateLimiter, rateLimiterData, combineKey);
        /*****************结束*****************/
        // 滑动窗口
        // state=slidingWindowRateState(rateLimiter,combineKey);
        log.warn("状态:" + state);
        try {
            if (!state) {
                //Xml格式返回
                if ("0".equals(resultType)) {
                    log.warn("访问过于频繁，请稍候再试");
                    //院感
                    if (limitResultType.getCode().equals(LimitResultType.NAME.getCode())) {
                        throw new XmlRRException(LimitResultType.NAME, "访问过于频繁，请稍候再试");
                    }
                    //LIS
                    if (limitResultType.getCode().equals(LimitResultType.NAME_PACS.getCode())) {
                        throw new XmlRRException(LimitResultType.NAME_PACS, ResultCodeEnum.SERVER_FAIL, "访问过于频繁，请稍候再试");
                    }
                }
            }
        } catch (XmlRRException e) {
            throw e;
        } catch (Exception e) {
            log.error("服务器限流异常，请稍候再试");
            throw new RuntimeException("服务器限流异常，请稍候再试");
        }


    }
   }

RateLimiterData.java

/**
 * 限流的参数
 */
@Data
public class RateLimiterData {


    /**
     * 请求数
     */
    private   int requestsMax = 0;

    /**
     * 请求时间
     */
    private long startTime =  0L;

}

RateLimiter.java

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
    /**
     * 限流key
     */
    String key() default "rate_limit:";

    /**
     * 限流时间,单位秒
     */
    int time() default 60;

    /**
     * 限流次数
     */
    int count() default 60;

    /**
     * 扣减数量
     */
    int deduction() default 1;

    /**
     * 返回类型 0 为xml 1为 json
     * @return
     */
    String resultType() default "0";




    /**
     * 限流类型
     */
    LimitType limitType() default LimitType.DEFAULT;

    /**
     * 厂商返回类型
     * @return
     */
    LimitResultType limitResultType() default LimitResultType.NAME_DEFAULT;
}

结束

• 固定窗口算法:实现简单，性能高，有明显的临界问题，瞬时流量可达到阈值的2倍。
• 滑动窗口算法:消除了临界问题，但是当流量到达阈值时会瞬间掐断流量，导致流量不够平滑。
• 漏桶算法:想要达到限流的目的，又不会掐断流量，使得流量更加平滑,但是因为是固定速率流出水,无法处理瞬时利流量,不能很快处理请求。
• 令牌算法:处理突发流量，同时保证一定程度的平滑流动。

---

此次以上就是限流的四种算法,具体使用那种算法需要根据自身的业务进行选择,如果是分布式的话推荐使用redis+Lua脚本来实现市面上也有其他的限流工具，
如:基于令牌桶实现的限流组件Guava,基于滑动窗口实现的分布式服务架构的流量控制框架Sentinel。

Lua脚本

令牌算法的脚本

-- 令牌总数
local capacity = tonumber(ARGV[1])
-- 令牌产生是速率(**个个数/秒)
local rate = tonumber(ARGV[2])
-- 访问时间
local lastTime = tonumber(ARGV[3])
-- 剩余的令牌数量
local residueTokenNum= 0
-- 令牌桶的过期时间(即key的过期时间)
local resetToken = tonumber(ARGV[4])
-- 消耗的数量
local consume=1
-- 获得桶令牌数据
local token = redis.call('hgetall', KEYS[1])
-- 当前桶未初始化,则进行初始化令牌桶
if table.maxn(token) == 0 then
-- 每次消耗一个令牌
local residueTokenNum= tonumber(ARGV[1]) - consume
-- 将初始化的数据存入
redis.call('hmset', KEYS[1], 'capacity', capacity, 'rate', rate,  'lastTime', lastTime,'residueTokenNum', residueTokenNum)
-- 设置过期时间
redis.call('pexpire',  KEYS[1], tonumber(ARGV[4]))
else
-- 获得已经存入的数据
local bucket = redis.call('HMGET', KEYS[1], 'capacity', 'rate', 'lastTime','residueTokenNum')

capacity = bucket[1]
rate =bucket[2]
lastTime = bucket[3]
residueTokenNum= bucket[4]
-- 传入的时间数据
local nowTime = tonumber(ARGV[3])
-- 生成的令牌数
local generateTokenNum = tonumber(((nowTime - lastTime)/1000)* rate)
-- 总剩余数量=生成数量+原剩余的数量
residueTokenNum = generateTokenNum + residueTokenNum
--  (桶的限制数量,总生成的数量) => 超出桶最大令牌数量则进行丢弃
residueTokenNum = math.min(capacity, residueTokenNum)
-- 剩余的总数量大于等于消耗的数量
if residueTokenNum >= 1 then
-- 剩余的数量=剩余的总数量-去消耗的令牌数量
residueTokenNum = residueTokenNum - consume
-- 重置参数
redis.call('hmset', KEYS[1], 'capacity', capacity, 'rate', rate, 'lastTime', nowTime,'residueTokenNum', residueTokenNum)
-- 返回剩余的数量
return residueTokenNum
end
return 0
end