美文网首页
Guava RateLimiter限流源码解析

Guava RateLimiter限流源码解析

作者: majinliang123 | 来源:发表于2019-03-27 17:26 被阅读0次

    RateLimiter的设计思想

    RateLimiter最大的特点是它可以以一个稳定的速率让线程通过。当线程过多的时候,它可以通过sleep使线程暂停,从而达到强制控制线程通过速度的目标。

    控制QPS最简单的方法是可以通过保留上个线程通过的时间,从而可以计算下次请求通过时间。但是,这样RateLimiter只知道最后一个线程通过的时间。如果RateLimiter长时间没有使用,然后一个请求到达并通过,这时RateLimiter会立即忘记在过去很长时间中,RateLimiter并没有被充分利用,这会导致RateLimiter利用率不足。由于过去未充分利用RateLimiter,这可能意味着过剩的资源,RateLimiter应该加速一段时间,从而充分利用过剩的资源。

    另一方面,过去资源未充分利用可能意味着它的缓存过期,在预热阶段发生很很消耗性能的操作,例如在Executors.newCachedThreadPool所创建的线程池中,线程超过1分钟未之后,就会被销毁,一个新的任务需要被执行时,需要重新创建线程。

    RateLimiter通过记录下个线程可以获取到令牌的时间,从而达到控制线程通过速率的目的。

    Guava中RateLimiter有两种实现SmoothBursty和SmoothWarmingUp,这里主要介绍SmoothBursty。

    重要属性介绍

    maxBurstSeconds:通过maxBurstSeconds来控制保存的令牌数量。如果一个RateLimiter是2QPS,maxBurstSeconds是10秒,这样,我们可以最多保存2*10=20个令牌。
    storedPermits:当前令牌桶中令牌的个数。
    maxPermits:令牌桶中做多存储令牌的数量。
    stableIntervalMicros:生成令牌的时间间隔。例如,每秒可以生成5个令牌,那stableIntervalMicros是200。
    nextFreeTickerMicros:生成下个令牌的时间。

    RateLimiter的创建

    在RateLimiter中,通过它的静态方法创建SmoothBursty

      public static RateLimiter create(double permitsPerSecond) {
        return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
      }
      @VisibleForTesting
      static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
        RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
        rateLimiter.setRate(permitsPerSecond);
        return rateLimiter;
      }
    
        SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {
          super(stopwatch);
          this.maxBurstSeconds = maxBurstSeconds;
        }
    

    stopwatch是控制有关时间和sleep线程的。
    maxBurstSeconds是用来控制保存令牌的数量的。

      public final void setRate(double permitsPerSecond) {
        checkArgument(
            permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
        synchronized (mutex()) {
          doSetRate(permitsPerSecond, stopwatch.readMicros());
        }
      }
    

    mutex()通过双检查锁始终返回一个同一个单例对象,从而使方法doSetRate()的线程安全。
    stopwatch.readMicros()读取了当前时间。

      @Override
      final void doSetRate(double permitsPerSecond, long nowMicros) {
        resync(nowMicros);
        double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
        this.stableIntervalMicros = stableIntervalMicros;
        doSetRate(permitsPerSecond, stableIntervalMicros);
      }
    

    resync方法主要用于同步RateLimiter中storedPermits和nextFreeTicketMicros.
    stableIntervalMicros是生产一个令牌的所使用的时间,比如permitsPerSecond=0.5(每秒钟生产0.5个令牌),那生产一个令牌的时间是2秒(2000毫秒)。
    doSetRate通过计算,计算出storedPermits和maxPermits的值。
    以上方法导致最终的结果就是storedPermits=0,maxPermits=0,nextFreeTicketMicros是当前的时间。

      void resync(long nowMicros) {
        // if nextFreeTicket is in the past, resync to now
        if (nowMicros > nextFreeTicketMicros) {
          double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
          storedPermits = min(maxPermits, storedPermits + newPermits);
          nextFreeTicketMicros = nowMicros;
        }
      }
    

    当RateLimiter初始化的时候,nextFreeTicketMicros=0。coolDownIntervalMicros方法返回stableIntervalMicros,在RateLimiter初始化的时候,stableIntervalMicros=0,所以newPermits是Infinity。在RateLimiter初始化的时候,maxPermits=0,所以storedPermits=0,表示刚初始化的时候,RateLimiter中没有缓存的令牌。
    resync还将nextFreeTicketMicros更新为当前的时间。

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

    当RateLimiter初始化的时候,会走else语句,因为oldMaxPermits是0,导致storedPermits=0。

    RateLimiter的使用

      @CanIgnoreReturnValue
      public double acquire(int permits) {
        long microsToWait = reserve(permits);
        stopwatch.sleepMicrosUninterruptibly(microsToWait);
        return 1.0 * microsToWait / SECONDS.toMicros(1L);
      }
    

    reserve方法会计算出当前线程下次启动需要等待的时间,然后通过sleep将线程暂停,当线程重新启动的时候,返回线程暂停的时间。

    final long reserve(int permits) {
        checkPermits(permits);
        synchronized (mutex()) {
          return reserveAndGetWaitLength(permits, stopwatch.readMicros());
        }
    }
    
    final long reserveAndGetWaitLength(int permits, long nowMicros) {
        long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
        return max(momentAvailable - nowMicros, 0);
    }
    
    @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;
    }
    

    reserveEarliestAvailable主要用于计算当前线程获取到令牌的时间,在计算的过程中会考虑缓存的令牌。
    还会计算下个线程获取到令牌的时间,并赋给nextFreeTicketMicros。

    相关文章

      网友评论

          本文标题:Guava RateLimiter限流源码解析

          本文链接:https://www.haomeiwen.com/subject/kkynvqtx.html