Guava RateLimiter限流

時間  2019-11-07
標籤 guava ratelimiter 限流 欄目 Java 简体版
原文   原文鏈接

緩存，降級和限流是大型分佈式系統中的三把利劍。目前限流主要有漏桶和令牌桶兩種算法。java

  1. 緩存：緩存的目的是減小外部調用，提升系統響速度。俗話說："緩存是網站優化第必定律"。緩存又分爲本機緩存和分佈式緩存，本機緩存是針對當前JVM實例的緩存，能夠直接使用JDK Collection框架裏面的集合類或者諸如Google Guava Cache來作本地緩存；分佈式緩存目前主要有Memcached,Redis等。
  2. 降級：所謂降級是指在系統調用高峯時，優先保證咱們的核心服務，對於非核心服務能夠選擇將其關閉以保證核心服務的可用。例如在淘寶雙11時，支付功能是核心，其餘諸如用戶中心等非核心功能能夠選擇降級，優先保證交易。
  3. 限流：任何系統的性能都有一個上限，當併發量超過這個上限以後，可1能會對系統形成毀滅性地打擊。所以在任什麼時候刻咱們都必須保證系統的併發請求數量不能超過某個閾值，限流就是爲了完成這一目的。

限流之漏桶算法

漏桶算法的示意圖以下：git

漏桶算法

漏桶算法能夠將系統處理請求限定到恆定的速率，當請求過載時，漏桶將直接溢出。漏桶算法假定了系統處理請求的速率是恆定的，可是在現實環境中，每每咱們的系統處理請求的速率不是恆定的。漏桶算法沒法解決系統突發流量的狀況。github

限流之令牌桶算法

令牌桶算法相對漏桶算法的優點在於能夠處理系統的突發流量，其算法示意圖以下所示：算法

令牌桶算法

令牌桶有必定的容量（capacity），後臺服務向令牌桶中以恆定的速率放入令牌（token），當令牌桶中的令牌數量超過capacity以後，多餘的令牌直接丟棄。當一個請求進來時，須要從桶中拿到N個令牌，若是可以拿到則繼續後面的處理流程，若是拿不到，則當前線程能夠選擇阻塞等待桶中的令牌數量夠本次請求的數量或者不等待直接返回失敗。segmentfault

Guava RateLimiter限流

Guava RateLimiter是一個谷歌提供的限流工具，RateLimiter基於令牌桶算法，能夠有效限定單個JVM實例上某個接口的流量。緩存

RateLimiter使用的一個例子

import com.google.common.util.concurrent.RateLimiter;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RateLimiterExample {

    public static void main(String[] args) throws InterruptedException {
        // qps設置爲5，表明一秒鐘只容許處理五個併發請求
        RateLimiter rateLimiter = RateLimiter.create(5);
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        int nTasks = 10;
        CountDownLatch countDownLatch = new CountDownLatch(nTasks);
        long start = System.currentTimeMillis();
        for (int i = 0; i < nTasks; i++) {
            final int j = i;
            executorService.submit(() -> {
                rateLimiter.acquire(1);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                System.out.println(Thread.currentThread().getName() + " gets job " + j + " done");
                countDownLatch.countDown();
            });
        }
        executorService.shutdown();
        countDownLatch.await();
        long end = System.currentTimeMillis();
        System.out.println("10 jobs gets done by 5 threads concurrently in " + (end - start) + " milliseconds");
    }
}

複製代碼

輸出結果:bash

pool-1-thread-1 gets job 0 done
pool-1-thread-2 gets job 1 done
pool-1-thread-3 gets job 2 done
pool-1-thread-4 gets job 3 done
pool-1-thread-5 gets job 4 done
pool-1-thread-6 gets job 5 done
pool-1-thread-7 gets job 6 done
pool-1-thread-8 gets job 7 done
pool-1-thread-9 gets job 8 done
pool-1-thread-10 gets job 9 done
10 jobs gets done by 5 threads concurrently in 2805 milliseconds
複製代碼

上面例子中咱們提交10個工做任務，每一個任務大概耗時1000微秒，開啓10個線程，而且使用RateLimiter設置了qps爲5，一秒內只容許五個併發請求被處理，雖然有10個線程，可是咱們設置了qps爲5，一秒以內只能有五個併發請求。咱們預期的總耗時大概是2000微秒左右，結果爲2805和預期的差很少。併發

RateLimiter

RateLimiter基於令牌桶算法，它的核心思想主要有：框架

  1. 響應本次請求以後，動態計算下一次能夠服務的時間，若是下一次請求在這個時間以前則須要進行等待。SmoothRateLimiter 類中的 nextFreeTicketMicros 屬性表示下一次能夠響應的時間。例如，若是咱們設置QPS爲1，本次請求處理完以後，那麼下一次最先的可以響應請求的時間一秒鐘以後。
  2. RateLimiter 的子類 SmoothBursty 支持處理突發流量請求，例如，咱們設置QPS爲1，在十秒鐘以內沒有請求，那麼令牌桶中會有10個（假設設置的最大令牌數大於10）空閒令牌，若是下一次請求是 acquire(20) ，則不須要等待20秒鐘，由於令牌桶中已經有10個空閒的令牌。SmoothRateLimiter 類中的 storedPermits 就是用來表示當前令牌桶中的空閒令牌數。
  3. RateLimiter 子類 SmoothWarmingUp 不一樣於 SmoothBursty ，它存在一個「熱身」的概念。它將 storedPermits 分紅兩個區間值：[0, thresholdPermits) 和 [thresholdPermits, maxPermits]。當請求進來時，若是當前系統處於"cold"的狀態，從 [thresholdPermits, maxPermits] 區間去拿令牌，所須要等待的時間會長於從區間 [0, thresholdPermits) 拿相同令牌所須要等待的時間。當請求增多，storedPermits 減小到 thresholdPermits 如下時，此時拿令牌所須要等待的時間趨於穩定。這也就是所謂「熱身」的過程。這個過程後面會詳細分析。

RateLimiter主要的類的類圖以下所示：分佈式

RateLimiter類圖

RateLimiter 是一個抽象類，SmoothRateLimiter 繼承自 RateLimiter，不過 SmoothRateLimiter 仍然是一個抽象類，SmoothBursty 和 SmoothWarmingUp 纔是具體的實現類。

SmoothRateLimiter主要屬性

SmoothRateLimiter 是抽象類，其定義了一些關鍵的參數，咱們先來看一下這些參數：

/**
* The currently stored permits.
*/
double storedPermits;

/**
* The maximum number of stored permits.
*/
double maxPermits;

/**
* The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits
* per second has a stable interval of 200ms.
*/
double stableIntervalMicros;

/**
* The time when the next request (no matter its size) will be granted. After granting a request,
* this is pushed further in the future. Large requests push this further than small requests.
*/
private long nextFreeTicketMicros = 0L; // could be either in the past or future
複製代碼

storedPermits 代表當前令牌桶中有多少令牌。maxPermits 表示令牌桶最大令牌數目，storedPermits 的取值範圍爲：[0, maxPermits]。stableIntervalMicros 等於 1/qps，它表明系統在穩按期間，兩次請求之間間隔的微秒數。例如：若是咱們設置的 qps 爲5，則 stableIntervalMicros 爲200ms。nextFreeTicketMicros 表示系統處理完當前請求後，下一次請求被許可的最短微秒數，若是在這以前有請求進來，則必須等待。

當咱們設置了 qps 以後，須要計算某一段時間系統可以生成的令牌數目，那麼怎麼計算呢？一種方式是開啓一個後臺任務去作，可是這樣代價未免有點大。RateLimiter 中採起的是惰性計算方式：在每次請求進來的時候先去計算上次請求和本次請求之間應該生成多少個令牌。

SmoothBursty

建立

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 */);  // maxBurstSeconds 用於計算 maxPermits
    rateLimiter.setRate(permitsPerSecond); // 設置生成令牌的速率
    return rateLimiter;
}
複製代碼

SmoothBursty 的 maxBurstSeconds 構造函數參數主要用於計算 maxPermits ：maxPermits = maxBurstSeconds * permitsPerSecond;。

咱們再看一下 setRate 的方法，RateLimiter 中 setRate 方法最終後調用 doSetRate 方法，doSetRate 是一個抽象方法，SmoothRateLimiter 抽象類中覆寫了 RateLimiter 的 doSetRate 方法：

// SmoothRateLimiter類中的doSetRate方法，覆寫了 RateLimiter 類中的 doSetRate 方法，此方法再委託下面的 doSetRate 方法作處理。
@Override
final void doSetRate(double permitsPerSecond, long nowMicros) {
    resync(nowMicros);
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    this.stableIntervalMicros = stableIntervalMicros;
    doSetRate(permitsPerSecond, stableIntervalMicros);
}

// SmoothBursty 和 SmoothWarmingUp 類中覆寫此方法
abstract void doSetRate(double permitsPerSecond, double stableIntervalMicros);

// SmoothBursty 中對 doSetRate的實現
@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; } } 複製代碼

resync方法

SmoothRateLimiter 類的 doSetRate方法中咱們着重看一下 resync 這個方法：

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;
    }
}
複製代碼

resync 方法就是 RateLimiter 中惰性計算 的實現。每一次請求來的時候，都會調用到這個方法。這個方法的過程大體以下：

  1. 首先判斷當前時間是否是大於 nextFreeTicketMicros ，若是是則表明系統已經"cool down"， 這兩次請求之間應該有新的 permit 生成。
  2. 計算本次應該新添加的 permit 數量，這裏分式的分母是 coolDownIntervalMicros 方法，它是一個抽象方法。在 SmoothBursty 和 SmoothWarmingUp 中分別有不一樣的實現。SmoothBursty 中返回的是 stableIntervalMicros 也便是 1 / QPS。coolDownIntervalMicros 方法在 SmoothWarmingUp 中的計算方式爲warmupPeriodMicros / maxPermits，warmupPeriodMicros 是 SmoothWarmingUp 的「預熱」時間。
  3. 計算 storedPermits，這個邏輯比較簡單。
  4. 設置 nextFreeTicketMicros 爲 nowMicros。

tryAcquire方法

tryAcquire 方法用於嘗試獲取若干個 permit，此方法不會等待，若是獲取失敗則直接返回失敗。canAcquire 方法用於判斷當前的請求可否經過：

public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
    long timeoutMicros = max(unit.toMicros(timeout), 0);
    checkPermits(permits);
    long microsToWait;
    synchronized (mutex()) {
        long nowMicros = stopwatch.readMicros();
        if (!canAcquire(nowMicros, timeoutMicros)) { // 首先判斷當前超時時間以內請求可否被知足，不能知足的話直接返回失敗
            return false;
        } else {
            microsToWait = reserveAndGetWaitLength(permits, nowMicros); // 計算本次請求須要等待的時間，核心方法
        }
    }
    stopwatch.sleepMicrosUninterruptibly(microsToWait);
    return true;
}

final long reserveAndGetWaitLength(int permits, long nowMicros) {
    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
    return max(momentAvailable - nowMicros, 0);
}

private boolean canAcquire(long nowMicros, long timeoutMicros) {
    return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros;
}

final long queryEarliestAvailable(long nowMicros) {
    return nextFreeTicketMicros;
}
複製代碼

canAcquire 方法邏輯比較簡單，就是看 nextFreeTicketMicros 減去 timeoutMicros 是否小於等於 nowMicros。若是當前需求能被知足，則繼續往下走。

接着會調用 SmoothRateLimiter 類的 reserveEarliestAvailable 方法，該方法返回當前請求須要等待的時間。改方法在 acquire 方法中也會用到，咱們來着重分析這個方法。

reserveEarliestAvailable方法

// 計算本次請求須要等待的時間
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {

    resync(nowMicros); // 本次請求和上次請求之間間隔的時間是否應該有新的令牌生成，若是有則更新 storedPermits
    long returnValue = nextFreeTicketMicros;
    
    // 本次請求的令牌數 requiredPermits 由兩個部分組成：storedPermits 和 freshPermits，storedPermits 是令牌桶中已有的令牌
    // freshPermits 是須要新生成的令牌數
    double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
    double freshPermits = requiredPermits - storedPermitsToSpend;
    
    // 分別計算從兩個部分拿走的令牌各自須要等待的時間，而後總和做爲本次請求須要等待的時間，SmoothBursty 中從 storedPermits 拿走的部分不須要等待時間
    long waitMicros =
        storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
            + (long) (freshPermits * stableIntervalMicros);
            
    // 更新 nextFreeTicketMicros，這裏更新的實際上是下一次請求的時間，是一種「預消費」
    this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
    
    // 更新 storedPermits
    this.storedPermits -= storedPermitsToSpend;
    return returnValue;
}

/**
* Translates a specified portion of our currently stored permits which we want to spend/acquire,
* into a throttling time. Conceptually, this evaluates the integral of the underlying function we
* use, for the range of [(storedPermits - permitsToTake), storedPermits].
*
* <p>This always holds: {@code 0 <= permitsToTake <= storedPermits}
*/
abstract long storedPermitsToWaitTime(double storedPermits, double permitsToTake);
複製代碼

上面的代碼是 SmoothRateLimiter 中的具體實現。其主要有如下步驟：

  1. resync，這個方法以前已經分析過，這裏再也不贅述。其主要用來計算當前請求和上次請求之間這段時間須要生成新的 permit 數量。
  2. 對於 requiredPermits ，RateLimiter 將其分爲兩個部分：storedPermits 和 freshPermits。storedPermits 表明令牌桶中已經存在的令牌，能夠直接拿出來用，freshPermits 表明本次請求須要新生成的 permit 數量。
  3. 分別計算 storedPermits 和 freshPermits 拿出來的部分的令牌數所須要的時間，對於 freshPermits 部分的時間比較好計算：直接拿 freshPermits 乘以 stableIntervalMicros 就能夠獲得。而對於須要從 storedPermits 中拿出來的部分則計算比較複雜，這個計算邏輯在 storedPermitsToWaitTime 方法中實現。storedPermitsToWaitTime 方法在 SmoothBursty 和 SmoothWarmingUp 中有不一樣的實現。storedPermitsToWaitTime 意思就是表示當前請求從 storedPermits 中拿出來的令牌數須要等待的時間，由於 SmoothBursty 中沒有「熱身」的概念， storedPermits 中有多少個就能夠用多少個，不須要等待，所以 storedPermitsToWaitTime 方法在 SmoothBursty 中返回的是0。而它在 SmoothWarmingUp 中的實現後面會着重分析。
  4. 計算到了本次請求須要等待的時間以後，會將這個時間加到 nextFreeTicketMicros 中去。最後從 storedPermits 減去本次請求從這部分拿走的令牌數量。
  5. reserveEarliestAvailable 方法返回的是本次請求須要等待的時間，該方法中算出來的 waitMicros 按理來講是應該做爲返回值的，可是這個方法返回的倒是開始時的 nextFreeTicketMicros ，而算出來的 waitMicros 累加到 nextFreeTicketMicros 中去了。這裏其實就是「預消費」，讓下一次消費來爲本次消費來「買單」。

acquire方法

acquire 方法沒有等待超時的概念，會一直阻塞直到知足本次請求。

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);
}  
  
abstract long reserveEarliestAvailable(int permits, long nowMicros);  
複製代碼

acquire 方法最終仍是經過 reserveEarliestAvailable 方法來計算本次請求須要等待的時間。這個方法上面已經分析過了，這裏就再也不過多闡述。

SmoothWarmingUp

SmoothWarmingUp 相對 SmoothBursty 來講主要區別在於 storedPermitsToWaitTime 方法。其餘部分原理和 SmoothBursty 相似。

建立

SmoothWarmingUp 是 SmoothRateLimiter 的子類，它相對於 SmoothRateLimiter 多了幾個屬性：

static final class SmoothWarmingUp extends SmoothRateLimiter {
    private final long warmupPeriodMicros;
    /**
     * The slope of the line from the stable interval (when permits == 0), to the cold interval
     * (when permits == maxPermits)
     */
    private double slope;
    private double thresholdPermits;
    private double coldFactor;
    ...
}
複製代碼

這四個參數都是和 SmoothWarmingUp 的「熱身」（warmup）機制相關。warmup 能夠用以下的圖來表示：

*          ^ 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
複製代碼

上圖中橫座標是當前令牌桶中的令牌 storedPermits，前面說過 SmoothWarmingUp 將 storedPermits 分爲兩個區間：[0, thresholdPermits) 和 [thresholdPermits, maxPermits]。縱座標是請求的間隔時間，stableInterval 就是 1 / QPS，例如設置的 QPS 爲5，則 stableInterval 就是200ms，coldInterval = stableInterval * coldFactor，這裏的 coldFactor "hard-code"寫死的是3。

當系統進入 cold 階段時，圖像會向右移，直到 storedPermits 等於 maxPermits；當系統請求增多，圖像會像左移動，直到 storedPermits 爲0。

storedPermitsToWaitTime方法

上面"矩形+梯形"圖像的面積就是 waitMicros 也便是本次請求須要等待的時間。計算過程在 SmoothWarmingUp 類的 storedPermitsToWaitTime 方法中覆寫:

@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) { // 若是當前 storedPermits 超過 availablePermitsAboveThreshold 則計算從 超過部分拿令牌所須要的時間（圖中的 WARM UP PERIOD）
        // WARM UP PERIOD 部分計算的方法，這部分是一個梯形，梯形的面積計算公式是 「（上底 + 下底） * 高 / 2」
        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); // 計算出從 WARM UP PERIOD 拿走令牌的時間
        permitsToTake -= permitsAboveThresholdToTake; // 剩餘的令牌從 stable 部分拿
    }
    // measuring the integral on the left part of the function (the horizontal line)
    micros += (stableIntervalMicros * permitsToTake); // stable 部分令牌獲取花費的時間
    return micros;
}

// WARM UP PERIOD 部分 獲取相應令牌所對應的的時間
private double permitsToTime(double permits) {
    return stableIntervalMicros + permits * slope;
}
複製代碼

SmoothWarmingUp 類中 storedPermitsToWaitTime 方法將 permitsToTake 分爲兩部分，一部分從 WARM UP PERIOD 部分拿，這部分是一個梯形，面積計算就是（上底 + 下底）* 高 / 2。另外一部分從 stable 部分拿，它是一個長方形，面積就是 長 * 寬。最後返回兩個部分的時間總和。

總結

  1. RateLimiter中採用惰性方式來計算兩次請求之間生成多少新的 permit，這樣省去了後臺計算任務帶來的開銷。
  2. 最終的 requiredPermits 由兩個部分組成：storedPermits 和 freshPermits 。SmoothBursty 中 storedPermits 都是同樣的，不作區分。而 SmoothWarmingUp 類中將其分紅兩個區間：[0, thresholdPermits) 和 [thresholdPermits, maxPermits]，存在一個"熱身"的階段，thresholdPermits 是系統 stable 階段和 cold 階段的臨界點。從 thresholdPermits 右邊的部分拿走 permit 須要等待的時間更長。左半部分是一個矩形，由半部分是一個梯形。
  3. RateLimiter 可以處理突發流量的請求，採起一種"預消費"的策略。
  4. RateLimiter 只能用做單個JVM實例上接口或者方法的限流，不能用做全局流量控制。

參考資料

使用Guava RateLimiter限流以及源碼解析
Guava API文檔

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程