对限流RateLimiter解析

喜欢ヅ旅行 2021-09-15 23:28 412阅读 0赞

## 0. 写在最前（最后）的总结 ##

RateLimter实现了令牌桶的限流算法，从头看到尾后，发现内容还是比较简单的，可以举个例子来快速理解。

假设我们希望每秒最多发送5个请求，那么相当于每0.2秒发送一个。   
\- 当第一个请求发送后，记为开始，即0秒。   
\- 0.1秒时，来了第二个请求，这时候还没到第0.2秒（可以认为当前令牌数为0），那么我们不需要考虑别的，等到0.2秒就能执行，但是，必须知道，这时候，下次允许执行的请求时间应该是第0.4秒了。   
\- 假如第二个请求执行完后，到2.0秒还是没有新的请求到来，那么我们可以理解0.4秒到2秒的空闲时间就保存了5个令牌（最多不能超过5个，而且这个计算实际上是在新请求到来的时刻计算出来的）。   
\- 当2.1秒时，有新请求进来了   
如果它需要消耗3个请求，那么它可以立刻执行，然后存储令牌就变成了2个，而且如果此时再有新请求进来，可以立刻执行（smoothbursty），或者等待一个消耗3个存储令牌的积分时间执行（smoothwarmingup）。   
如果它需要消耗9个请求，那么它也可以立刻执行，然后存储令牌清空，而且如果此时再有新请求进来，需要等待4个令牌的时间0.8秒（smoothbursty），或者等待一个消耗5个存储令牌的积分时间执行加上等待4个令牌的时间（smoothwarmingup）。

需要注意的是，桶里的令牌只有在空闲时才会增加，如果正常每0.2秒来一个请求，那么桶里的令牌一直为0。   
空闲时的令牌最大可存储的数量5（smoothbursty），或者根据warmupPeriod调整（smoothwarmingup）。

## 1. SmoothRateLimiter的设计哲学 ##

RateLimiter是怎么设计的，为什么这样设计？

RateLimiter的主要特性是它的“稳定速率”，表示正常情况下允许的最大速率。这是根据“节流”输入请求来强制执行的，例如，对于一个请求而言，计算合适的节流时间，然后让改线程等待相应的时间。

最简单的维持QPS速率的方法是保持上次请求的时间戳，并且保证（1/QPS）秒的间隔。比如说，当QPS=5时，我们需要确保不会在上个请求的200ms内处理新的请求，这样就能维持想要的速率。如果一个新的请求到来，而上一个请求仅在100ms前，我们就需要再等待100ms。按照这样的速度，服务15个许可证（例如一个aquire(15)的请求）自然需要3秒钟。

对于这样一个RateLimiter，我们需要认识到它对过去有很肤浅的记忆：它只记得最后一个请求。如果这个RateLimiter有很长一段时间没有使用了，然后突然来了一个请求，那么它马上就会被执行。这个RateLimiter忘记了之前的“不饱和”状态。这个结果会导致“不饱和”或者溢出，取决于不使用期望速率的真实后果。

一方面，过去的“不饱和”可能意味着过剩的可用资源。然后，这个RateLimiter需要一定的时间提速，来充分利用这些资源。这一点很重要，这个在网络带宽受限的时候，过去的不饱和通常意味着“几乎空的缓存”，会被立即填满。

另一方面，过去的不饱和可能意味着，负责处理请求的服务器可能没有做好准备来处理之后的请求。比如，它的cache变得陈旧，并且请求可能触发更加昂贵的操作（一个极端的例子是一个服务器可能刚刚重启，它可能正在忙于处理自身的事情）。

为了解决这个问题，我们引入了一个额外的方面，对“过去的不饱和”状态，用“storePermits”（翻译成：存储许可，其实就是令牌）变量来建模。当没有不饱和时，这个变量为0，当有了非常大的不饱和情况时，它能够增长到最大允许存储许可数。所以，一个请求需要许可时，派发的许可来源于两方面：存储的许可（如果有的话），和新产生的许可

它是如何生效的呢？   
对于一个RateLimiter来说，它每秒产生一个token，随着RateLimiter闲置，存储许可每秒会增长1。也就是说，Ratelimiter闲着10秒，存储许可就达到10了（假设最大允许许可数≥10）。在这种情况下，一个请求acquire(3)到达。我们使用存储许可来为这个请求服务，然后存储许可降低到7。紧接着，假设一个acquire(10)的请求到来，我们使用所有的7个存储许可，然后剩下的3个许可，我们通过Ratelimiter产生新的许可。   
我们知道，如果Ratelimiter产生许可的速率是每秒一个token的话，那就需要三秒钟。那么，那7个存储许可怎么利用呢？正如上面解释的，没有唯一的答案。如果我们的关注点在处理不饱和，那么我们就希望存储许可被利用得比新产生许开要快，因为不饱和程度意味着可被利用的空闲资源；如果我们主要关注的是处理溢出，那么存储许开就要被利用得比新产生许可要慢。因此，针对不同情况，我们需要一个函数来表示存储许可与空闲时间的关系。

这个角色由storedPermitsToWaitTime(double storedPermits, double permitsToTake)来扮演。基础模型是一个连续函数，映射了存储许可（从0到最大允许存储许可）在 1/Rate 上的有效存储许可。“存储许可”基本上能够度量出未使用时间。Rate是“permits/time”，所以“1/rate”就是“time/permits”，所以和permit数量相乘就表示了时间 。例如，在这个函数上做积分（就是 storedPermitsToWaitTime （）的计算方式），对于给定数量的请求许可，等价于后续请求的最小间隔时间。这个在SmoothBursty和SmoothWarmingUp是不同的实现，SmoothBursty直接返回无需等待，SmoothWarmingUp则要计数出等待时间。

这里有个例子。如果storedPermits==10，然后我们需要3个permits，我们从存储许可中拿走3个，还剩7个。那么对于这消耗3个许可的时间，就需要调用storedPermitsToWaitTime(storedPermits = 10.0, permitsToTake = 3.0)进行计算节流时间，就是对7到10进行积分计算。   
使用积分保证了一个单独的acquire(3)等价于三个acquire(1)，或者一个acquire(2)和一个acquire(1)，不管使用怎样的积分函数。   
需要注意的是，对于这个积分函数，如果我们选择一个水平线，高度正好等于 1/QPS，那么这个函数就没啥作用了，因为它表示存储许可和新产生许可的速率是一致的。   
如果我们选择一个积分函数低于这个水平线，它表示我们减少了积分面积，也就是时间。因此，Ratelimiter在一段空闲后变得更加快速了。反之，如果我们选择一个积分函数高于这个水平线，那就意味着面积（时间）增大了，因此存储许可就比新产生许可要更加耗时，也就是Ratelimiter在一段空闲后变慢了。

最后一点是，如果RateLimiter 采用QPS＝1的限定速度，那么开销较大的acquire(100)请求到达时，它是没必要等到100s 才开始实际任务。我们可以先开始任务执行，并把未来的请求推后100s,这样我们就可以同时工作，同时生产需要的permits,而不是空等待permits生产够才开始工作。

**这里有一个重要的结论：RateLimiter不需要记住上个请求的时间,它只需要记住“希望下个请求到来的时间”即可。这样使得我们能够马上识别出，一个确切的超时时间（跟tryAcquire(timeout)相关）是否满足下一个计划时间点。 如果当前时间大于“期望的下个请求到来的时间”，那么这两个时间的差值就是Ratelimiter的未使用时间t，通过t\*QPS计算出storedPermits。**

这个方法就是 SmoothRateLimiter 中的reserveEarliestAvailable方法。

## 2. 源码解析 ##

我们先来看看SmoothRateLimiter类中的主要属性：

// 最大许可数，比如RateLimiter.create(2);表明最大的许可数就是2
    
        double maxPermits;
    
        // 当前剩余的许可数
    
        double storedPermits;
    
        // 固定的微秒周期
    
        double stableIntervalMicros;
    
        // 下一个空闲票据的时间点
    
        private long nextFreeTicketMicros = 0L;

因为RateLimiter需要保证许可被稳定连续的输出，比如每秒有5个许可，那么你获取许可的间隔时间是200毫秒，而stableIntervalMicros就是用来保存这个固定的间隔时间的，方便后面计算使用。

SmoothRateLimiter有两个子类：SmoothBursty和SmoothWarmingUp，从名字可以看出，SmoothWarmingUp类的实现带有预热功能，而SmoothBursty类是没有的，什么是预热功能呢？就好比缓存一样，当使用SmoothWarmingUp的实现时，不会在前几秒就给足全量的许可，就是说许可数会慢慢的增长，知道达到我们预定义的值，所以，SmoothRateLimiter会留下如下抽象方法交给其子类来实现，包括前文所述的关键方法storedPermitsToWaitTime()：

// 重设流量相关参数，需要子类来实现，不同子类参数不尽相同，比如SmoothWarmingUp肯定有增长比率相关参数
    
        void doSetRate(double permitsPerSecond, double stableIntervalMicros);
    
        // 计算生成这些许可数需要等待的时间
    
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake);
    
        // 返回许可冷却（间隔）时间
    
        double coolDownIntervalMicros();

**看完源码后发现，令牌桶算法其实不能望文生义，实现上并不是有个定时器在那里放令牌，然后消耗令牌，而是将空闲时间利用令牌的思想进行建模，实质上是计算并保留下次可执行请求的时间，并根据这个时间与请求到来的时间作比较，然后决定是否执行**

具体的实现可以继续看下面的源码。

### 2.1 以子类SmoothBursty为例 ###

抽象类RateLimiter的create方法返回的是一个“平滑突发”类型SmoothBursty的实例：

public static RateLimiter create(double permitsPerSecond) { 
            return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
        }
    
        @VisibleForTesting
        static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
            // 这里就默认设置了1秒钟，表示如果SmoothBursty没被使用，许可数能被保存1秒钟
            RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
            rateLimiter.setRate(permitsPerSecond);
            return rateLimiter;
        }

SmoothBursty的构造函数有两个入参，一个是SleepingStopwatch 实例，它是一个阻止睡眠观察器，可以理解成一个闹钟，如果你想睡十秒，它会在10秒钟内一直阻塞你（让你睡觉），到了十秒钟，它将对你放行（唤醒你），另一个参数是，每秒有几个许可，描述了限速的大小。

我们直接看获取令牌的阻塞实现acquire：

public double acquire(int permits) {
            // 预定permits个许可供未来使用，并返回等待这些许可需要的时间（微秒数）
            long microsToWait = reserve(permits);
            // 等待指定的时间
            stopwatch.sleepMicrosUninterruptibly(microsToWait);
            // 返回消耗的时间
            return 1.0 * microsToWait / SECONDS.toMicros(1L);
        }

思路比较清楚，关键在于计算需要等待的时间，继续往下看

/**
       * 向RateLimiter预定许可数，返回预定许可可以被消费的时间
       */
        final long reserve(int permits) {
            // permits的值必须大于0
            checkPermits(permits);
            // 这里通过一个对象锁进行同步
            synchronized (mutex()) {
                // 预定permits数目的许可，并返回调用方需要等待的时间
                return reserveAndGetWaitLength(permits, stopwatch.readMicros());
            }
        }

当拿到mutexDoNotUseDirectly锁后，我们看看reserveAndGetWaitLength如何计算等待时间

final long reserveAndGetWaitLength(int permits, long nowMicros) {
            //reserveEarliestAvailable()由RateLimiter的子类SmoothRateLimiter实现
            //计算可执行时间点
            long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
            //（等待时间 = 可执行时间点 - 当前时间点） 或者 马上执行
            return max(momentAvailable - nowMicros, 0);
        }

可以看出，获取许可需要等待的时间是由reserveEarliestAvailable方法来实现的，它是RateLimiter的抽象方法，由其抽象子类SmoothRateLimiter来实现。

现在，继续看reserveEarliestAvailable方法如何实现

/**
        * 返回可执行时间点
        */
        @Override
        final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
            // 基于当前时间nowMicros来更新storedPermits和nextFreeTicketMicros
            resync(nowMicros);
            long returnValue = nextFreeTicketMicros;
            // 从需要申请的许可数和当前可用的许可数中找到最小值
            double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
            double freshPermits = requiredPermits - storedPermitsToSpend;
            long waitMicros =
                storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
                    + (long) (freshPermits * stableIntervalMicros);
            // 更新下一个票据的等待时间
            this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
            // 更新剩余可用许可数
            this.storedPermits -= storedPermitsToSpend;
            return returnValue;
        }
    
        /** Updates {@code storedPermits} and {@code nextFreeTicketMicros} based on the current time. */
        void resync(long nowMicros) {
            // 如果下个可执行时间点大于当前时间，那么就继续等待，不需要更新
            // 反之，说明又空闲了一段时间，所以存储许可的数量肯定又增多了，可执行时间就是当前时间
            if (nowMicros > nextFreeTicketMicros) {
                //coolDownIntervalMicros()SmoothBurst中采用固定的间隔，SmoothWarmingUp中则跟warmupPeriodMicros有关
                //计算出新的空闲时间内产生的存储许可数量
                double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
                storedPermits = min(maxPermits, storedPermits + newPermits);
                nextFreeTicketMicros = nowMicros;
            }
        }

在SmoothBurst中的storedPermitsToWaitTime()非常简单，直接返回了0

@Override
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
          return 0L;
        }

也就是说，对于原本存储的许可直接消耗，不设定另外的等待时间，只用考虑freshPermit的生产时间。   
也就是 waitMicros = 0 + freshPermits \* stableIntervalMicros

### 2.2 子类SmoothWarmingUp的关键不同点 ###

create

create的时候初始化了一些参数，包括最大令牌数、冷却时间等。   
SmoothWarmingUp子类创建的时候需要说明warmupPeriod

public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
            checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
            //手动设置了coldfactor
            return create(
                permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
        }
    
        @VisibleForTesting
        static RateLimiter create(
            double permitsPerSecond,long warmupPeriod,TimeUnit unit, double coldFactor,        SleepingStopwatch stopwatch) {
            RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
            rateLimiter.setRate(permitsPerSecond);
            return rateLimiter;
        }

2.1中说明的create方法中，也有setRate()方法，这个方法里面检查了参数的合法性，然后就是调用doSetRate()方法，这是SmoothRateLimiter留下的三个交给其子类来实现抽象方法之一，我们对它们作个比较。

/**
        * 这里做的事情就比较简单，就是计算下最大允许存储许可数量和初始化存储许可数量
        */
        @Override
        void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
          double oldMaxPermits = this.maxPermits;
          maxPermits = maxBurstSeconds * permitsPerSecond;
          if (oldMaxPermits == Double.POSITIVE_INFINITY) {
            // if we don't special-case this, we would get storedPermits == NaN, below
            storedPermits = maxPermits;
          } else {
            storedPermits =
                (oldMaxPermits == 0.0)
                    ? 0.0 // initial state
                    : storedPermits * maxPermits / oldMaxPermits;
          }
        }
    
    
        @Override
        void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
          double oldMaxPermits = maxPermits;
          double coldIntervalMicros = stableIntervalMicros * coldFactor;
          thresholdPermits = 0.5 * warmupPeriodMicros / stableIntervalMicros;
          maxPermits =
              thresholdPermits + 2.0 * warmupPeriodMicros / (stableIntervalMicros + coldIntervalMicros);
          slope = (coldIntervalMicros - stableIntervalMicros) / (maxPermits - thresholdPermits);
          if (oldMaxPermits == Double.POSITIVE_INFINITY) {
            // if we don't special-case this, we would get storedPermits == NaN, below
            storedPermits = 0.0;
          } else {
            storedPermits =
                (oldMaxPermits == 0.0)
                    ? maxPermits // initial state is cold
                    : storedPermits * maxPermits / oldMaxPermits;
          }
        }

两者的初始storePermit是不一样的。   
然后这里的算法在javadoc中给出了详细的图文说明（**简单的数学推导，有兴趣可以看看，反正就是神奇的自定义参数**）

/**
       * This implements the following function where coldInterval = coldFactor * stableInterval.
       *
       * <pre>
       *          ^ throttling
       *          |
       *    cold  +                  /
       * interval |                 /.
       *          |                / .
       *          |               /  .   ← "warmup period" is the area of the trapezoid between
       *          |              /   .     thresholdPermits and maxPermits
       *          |             /    .
       *          |            /     .
       *          |           /      .
       *   stable +----------/  WARM .
       * interval |          .   UP  .
       *          |          . PERIOD.
       *          |          .       .
       *        0 +----------+-------+--------------→ storedPermits
       *          0 thresholdPermits maxPermits
       * </pre>
       *
       * 对这个函数进行详细说明前，我们需要了解以下基本信息：
    
       * <ol>
       *   <li>RateLimiter的状态 (storedPermits数量)是图中的垂直线
       *   <li>当RateLimiter进入空闲后，这个线会向右移动(直到 maxPermits)
       *   <li>当RateLimiter 被使用后, 这个线会向左移动(直到 zero), 因为有存储的许可，我们就会先从存储许可中拿
       *   <li>当不使用的时候，我们的线向右移动，保持恒定的速率。这个rate是 maxPermits / warmupPeriod. 
       *       这样就能保证从0移动到maxPermits的时间等于warmupPeriod
       *   <li>当使用许可的时候，它所需要的时间，如设计哲学中的解释，它等于在X-K到X上对该函数的积分，假定我们
       *       需要消耗X个许可
       * </ol>
       *
       * <p>总的来说，向左移动K个permit（就是消耗K个permit）的时间就是这个函数上间隔为K的积分面积。
       *
       * <p>假设我们有一个饱和模型（百度说是理想的人为定义模型），从maxPermit到thresholdPermit的
       * 时间就等于warmupPeriod. 然后从thresholdPermit到0的时间等于warmupPeriod/2. 
       * (这里是warmupPeriod/2的原因是为了保持原始的实现行为，也就是coldFactor被硬编码为3.）
       *
       * <p>这里还需要去计算 thresholdsPermits 和 maxPermits.
       *
       * <ul>
       *   <li>从 thresholdPermits 到 0 的积分是 thresholdPermits * stableIntervals. 也就是 warmupPeriod/2. 
       *       这部分就是最大速率的消耗时间，也就是正常速度
       *       因此
       *       thresholdPermits = 0.5 * warmupPeriod / stableInterval
       *   <li>从 maxPermits 到 thresholdPermits 的积分是一个梯形面积，等于
       *        0.5 * (stableInterval + coldInterval) * (maxPermits - thresholdPermits). 
       *       等于 warmupPeriod
       *       这部分是热身限速阶段
       *       因此
       *       maxPermits = thresholdPermits + 2 * warmupPeriod / (stableInterval + coldInterval)
       * </ul>
       */

acquire()

acquire()基本过程是一致的，差别注意体现在对存储许可消耗的时间上。

@Override
        final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
            //第一处不同，对空闲时间对应生成新令牌数的计算不同
            resync(nowMicros);
            long returnValue = nextFreeTicketMicros;
            double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
            double freshPermits = requiredPermits - storedPermitsToSpend;
            //第二处不同，对消耗存储令牌的时间不同，体现在下次请求需要等待的时间上。
            long waitMicros =
                storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
                    + (long) (freshPermits * stableIntervalMicros);
    
            this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
            this.storedPermits -= storedPermitsToSpend;
            return returnValue;
        }

第一处不同：

void resync(long nowMicros) {
            if (nowMicros > nextFreeTicketMicros) {
                //计算出新的空闲时间内产生的存储许可数量
                double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
                storedPermits = min(maxPermits, storedPermits + newPermits);
                nextFreeTicketMicros = nowMicros;
            }
        }

coolDownIntervalMicros()SmoothBurst中采用固定的间隔stableIntervalMicros，SmoothWarmingUp中则跟warmupPeriodMicros有关,这个在上面的算法中有介绍到，很容易理解

@Override
        double coolDownIntervalMicros() {
          return warmupPeriodMicros / maxPermits;
        }

第二处不同：

//这里就是把存储的令牌马上消耗，不计入新的等待时间内。只要有令牌空闲，就能执行请求
        @Override
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
          return 0L;
        }
    
        //就是上面对warmupPeriod的计算，分为水平线部分和梯形部分
        @Override
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
          double availablePermitsAboveThreshold = storedPermits - thresholdPermits;
          long micros = 0;
          // measuring the integral on the right part of the function (the climbing line)
          if (availablePermitsAboveThreshold > 0.0) {
            double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake);
            // TODO(cpovirk): Figure out a good name for this variable.
            double length =
                permitsToTime(availablePermitsAboveThreshold)
                    + permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake);
            micros = (long) (permitsAboveThresholdToTake * length / 2.0);
            permitsToTake -= permitsAboveThresholdToTake;
          }
          // measuring the integral on the left part of the function (the horizontal line)
          micros += (long) (stableIntervalMicros * permitsToTake);
          return micros;
        }

所以就是对上面那个函数的代码实现。原理也很清楚了。   
That’s all !

## 3. 有意思的代码 ##

// Can't be initialized in the constructor because mocks don't call the constructor.
      @MonotonicNonNullDecl private volatile Object mutexDoNotUseDirectly;
    
      private Object mutex() {
        Object mutex = mutexDoNotUseDirectly;
        if (mutex == null) {
          synchronized (this) {
            mutex = mutexDoNotUseDirectly;
            if (mutex == null) {
              mutexDoNotUseDirectly = mutex = new Object();
            }
          }
        }
        return mutex;
      }