服務限流之Guava源碼解析

1、前言

在應用系統中有三把利器能夠保護系統，提升系統健壯性：緩存、限流、降級，它們分別在不一樣場景下使用，但最終但目的都是用來保護系統穩定運行。其各自的使用場景及用途以下：

1）緩存：提升系統響應速度，爲數據庫減壓，避免過多查詢數據庫；

2）限流：限制併發量，限制某一段時間只有指定數量的請求進入後臺服務器，遇到流量高峯期或者流量突增（流量尖刺）時，把流量速率限制在系統所能接受的合理範圍以內，不至於讓系統被高流量擊垮；

3）降級：當訪問某個服務出現問題或影響核心業務時，暫時屏蔽該服務，直到請求高峯事後或者服務正常時再打開。

2、經常使用限流算法

常見等限流算法主要有2種：令牌桶、漏桶。令牌桶算法是令牌以固定速率流入桶內，桶滿了則丟棄令牌，桶內沒有令牌時則阻塞，當桶內有令牌但不夠時，不會阻塞，容許預先消費，可解決流量突發狀況。漏桶法則容許以任意速率流入桶內，但流出速率是恆定的，桶滿了則丟棄令牌，桶內沒有令牌時則阻塞，它不能解決流量突發狀況。具體流程如圖所示：

1）令牌桶法

2）漏桶法

3、常見的限流場景

開發過程當中或多或少會接觸到服務限流，好比tomcat限制最大鏈接數、數據庫鏈接池限制鏈接數、nginx限制ip訪問數、秒殺、搶購等。這些都是經過限制一個時間窗口內的請求數，當達到設置的最大請求數後，會讓後續請求進入等待隊列或直接拒絕，防止系統過載。這些限流是怎麼作到的呢，或者說限流主要有哪些方式呢？不少人把限流概括成4種場景，其實也不外乎這4種場景，它們分別是：

1）限制總併發數或請求數：針對整個系統作攔截

2）限制接口的總併發或請求數：針對單個接口作攔截

3）限制接口每秒的請求數：針對整個接口作攔截，時間窗口爲1秒

4）平滑限流接口的請求數：平滑突發限流（SmoothBursty）、平滑預熱限流（SmoothWarmingUp）

4、Guava使用及原理解析

4.1 Guava使用

Guava是google提供的開源工具包，提供了限流工具類RateLimiter，該類基於令牌桶算法實現流量限制，只需在pom.xml文件引入guava依賴，使用起來很方便。

從執行結果能夠看到第一次獲取時等待時間爲0.0s，後續每次獲取令牌都需等待0.5秒，0.5表示每秒往容器裏平滑的放2個憑證，每0.5秒放入一個，0.5是怎麼得來的呢？其實就是1/2=0.5，而2便是RateLimiter.create(2)這個方法中的參數值。那麼Guava是怎麼實現的呢？

4.2 Guava實現原理

Guava的RateLimiter限流步驟主要分2步：初始化、獲取憑證，初始化主要是建立RateLimiter對象，並初始化對應的屬性值，其中比較重要的幾個屬性以下：

4.1.1 容器初始化

初始化過程會作以下賦值：storedPermits=0、nextFreeTicketMicros=當前時間、stableIntervalMicros=1/qps，具體流程以下：

static RateLimiter create(SleepingStopwatch stopwatch, double permitsPerSecond) {
  // 建立RateLimiter對象
  RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
  // 初始化storedPermits、nextFreeTicketMicros、stableIntervalMicros的值
  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) {
  // 計算storedPermits、nextFreeTicketMicros的值
  resync(nowMicros);
  // 計算每一個憑證流入的間隔時間
  double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
  this.stableIntervalMicros = stableIntervalMicros;
  doSetRate(permitsPerSecond, stableIntervalMicros);
}

 

void resync(long nowMicros) {
  // 若是當前時間可以獲取憑證，每次獲取憑證時都會計算下一次能獲取憑證的時間
  if (nowMicros > nextFreeTicketMicros) {
    // 容器裏的總憑證數，即result=容器中原有憑證數+該時間點應該往容器裏放多少憑證
    storedPermits = min(maxPermits,
        storedPermits
          + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());
    nextFreeTicketMicros = nowMicros;
  }
}

4.1.2 獲取憑證

public double acquire(int permits) {
  // 計算須要等待的時間
  long microsToWait = reserve(permits);
  // 線程阻塞指定時間（tryAcquire()方法未非阻塞方法）
  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);
  // 2個數取最大值，當存在預先消費或提早獲取憑證時momentAvailable>nowMicros
  return max(momentAvailable - nowMicros, 0);
}

 

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
  // 計算storedPermits、nextFreeTicketMicros當值
  resync(nowMicros);
  long returnValue = nextFreeTicketMicros;
  double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
  // 計算容器中還需刷新的憑證數，即預先消費時會存在requiredPermits>storedPermitsToSpend
  double freshPermits = requiredPermits - storedPermitsToSpend;
  // 計算下一次獲取憑證時需等待的時間，預先消費時該值爲freshPermits * stableIntervalMicros
  long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
      + (long) (freshPermits * stableIntervalMicros);

  try {
    // 從新計算nextFreeTicketMicros的值
    this.nextFreeTicketMicros = LongMath.checkedAdd(nextFreeTicketMicros, waitMicros);
  } catch (ArithmeticException e) {
    this.nextFreeTicketMicros = Long.MAX_VALUE;
  }
  // 減小容器中的憑證數
  this.storedPermits -= storedPermitsToSpend;
  return returnValue;
}

 

void resync(long nowMicros) {
  // 若是當前時間可以獲取憑證，每次獲取憑證時都會計算下一次能獲取憑證的時間 
  if (nowMicros > nextFreeTicketMicros) {
    // 容器裏的總憑證數，即result=容器中原有憑證數+該時間點應該往容器裏放多少憑證 
    storedPermits = min(maxPermits,
        storedPermits
          + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());
    nextFreeTicketMicros = nowMicros;
  }
}