Guava RateLimiter限流源码解析

2019-03-27  本文已影响0人  majinliang123

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。

上一篇下一篇

猜你喜欢

热点阅读