RateLimiter

• 演算法原理

  • RateLimiter 從概念上來講，速率限制器會在可配置的速率下分配許可證。如果必要的話，每個acquire() 會阻塞當前執行緒直到許可證可用後獲取該許可證。一旦獲取到許可證，不需要再釋放許可證。

• 使用案例

  • RateLimiter.create(10);
  • RateLimiter.acquire();

• 有兩個實現子類

  • SmoothBursty

    • acquire後立即開始

  • SmoothWarmingUp

    • 有一段warm up間隔，後再開始

原始碼解析(SmoothBursty)

建立令牌桶

* RateLimiter.create(permitsPerSecond);

    * permitsPerSecond : 每一秒發放的令牌數
 

核心引數

• maxBurstSeconds

  • 統計的時間範圍，預設為1，即每一秒內發放的令牌

• nextFreeTicketMicros

  • 下一個令牌可以獲取的時間點，單位微秒

• storedPermits

  • 已有的令牌數

• maxPermits

  • 最大的令牌數

  • maxPermits = permitsPerSecond * maxBurstSeconds

• stableIntervalMicros

  • 令牌之間的平均間隔時間，單位微秒

  • stableIntervalMicros = 1秒 / permitsPerSecond

原始碼分析

public static RateLimiter create(double permitsPerSecond) {
    return create(SleepingStopwatch.createFromSystemTimer(), permitsPerSecond);
}
static RateLimiter create(SleepingStopwatch stopwatch, double permitsPerSecond) {
    //設定時間範圍長度
    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
    rateLimiter.setRate(permitsPerSecond);
    return rateLimiter;
}
public final void setRate(double permitsPerSecond) {
    checkArgument(
        permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
    //互斥鎖
    synchronized (mutex()) {
      //設定令牌桶頻率
      doSetRate(permitsPerSecond, stopwatch.readMicros());
    }
}
final void doSetRate(double permitsPerSecond, long nowMicros) {
    //更新引數
    resync(nowMicros);
    //令牌平均間隔時間
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    this.stableIntervalMicros = stableIntervalMicros;
    doSetRate(permitsPerSecond, stableIntervalMicros);
}
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
  double oldMaxPermits = this.maxPermits;
  //最大令牌數
  maxPermits = maxBurstSeconds * permitsPerSecond;
  //設定令牌桶的已有令牌數，初始狀態時為0
  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;
  }
}
//更新引數，入參為當前時間
void resync(long nowMicros) {
    // 如果當前時間超過下一個令牌可獲取時間，更新引數，並且將下個令牌的可獲取時間更新為當前時間
    if (nowMicros > nextFreeTicketMicros) {
      //更新令牌桶裡儲存的令牌數，包括 nowMicros - nextFreeTicketMicros 時間範圍內補充的令牌數
      storedPermits = min(maxPermits,
          storedPermits
            + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());
      nextFreeTicketMicros = nowMicros;
    }
 

}

獲取令牌

• rateLimiter.acquire(int permits)

  • permits : 需要獲取的令牌數

原始碼分析

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

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

//獲取最近的一個可用令牌的時間
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);

    try {
        //更新下個可用令牌的可獲取時間,最近可用令牌的時間+需要等待的時間
        this.nextFreeTicketMicros = LongMath.checkedAdd(nextFreeTicketMicros, waitMicros);
    } catch (ArithmeticException e) {
        this.nextFreeTicketMicros = Long.MAX_VALUE;
    }
    //更新令牌桶裡的令牌數
    this.storedPermits -= storedPermitsToSpend;
    return returnValue;
}
long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
  return 0L;
}