限流算法之漏桶算法、令牌桶算法

昨天CodeReview的時候看到同時使用RateLimiter這個類用作QPS訪問限制.學習一下這個類.

RateLimiter是Guava的concurrent包下的一個用於限制訪問頻率的類.

1.限流

每個API接口都是有訪問上限的,當訪問頻率或者並發量超過其承受范圍時候,我們就必須考慮限流來保證接口的可用性或者降級可用性.即接口也需要安裝上保險絲,以防止非預期的請求對系統壓力過大而引起的系統癱瘓.

通常的策略就是拒絕多余的訪問,或者讓多余的訪問排隊等待服務,或者引流.

如果要准確的控制QPS,簡單的做法是維護一個單位時間內的Counter,如判斷單位時間已經過去,則將Counter重置零.此做法被認為沒有很好的處理單位時間的邊界,比如在前一秒的最后一毫秒里和下一秒的第一毫秒都觸發了最大的請求數,將目光移動一下,就看到在兩毫秒內發生了兩倍的QPS.

2.限流算法

常用的更平滑的限流算法有兩種:漏桶算法和令牌桶算法.

很多傳統的服務提供商如華為中興都有類似的專利,參考: http://www.google.com/patents/CN1536815A?cl=zh

2.1 漏桶算法

漏桶(Leaky Bucket)算法思路很簡單,水(請求)先進入到漏桶里,漏桶以一定的速度出水(接口有響應速率),當水流入速度過大會直接溢出(訪問頻率超過接口響應速率),然后就拒絕請求,可以看出漏桶算法能強行限制數據的傳輸速率.示意圖如下:

可見這里有兩個變量,一個是桶的大小,支持流量突發增多時可以存多少的水(burst),另一個是水桶漏洞的大小(rate)，偽代碼如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

double rate; // leak rate in calls/s   double burst; // bucket size in calls       long refreshTime; // time for last water refresh   double water; // water count at refreshTime       refreshWater() {   long now = getTimeOfDay();       //水隨着時間流逝,不斷流走,最多就流干到0.   water = max(0, water- (now - refreshTime)*rate);   refreshTime = now;   }       bool permissionGranted() {   refreshWater();   if (water < burst) { // 水桶還沒滿,繼續加1   water ++;   return true;   } else {   return false;   }   }

因為漏桶的漏出速率是固定的參數,所以,即使網絡中不存在資源沖突(沒有發生擁塞),漏桶算法也不能使流突發(burst)到端口速率.因此,漏桶算法對於存在突發特性的流量來說缺乏效率.

2.2 令牌桶算法

令牌桶算法(Token Bucket)和 Leaky Bucket 效果一樣但方向相反的算法,更加容易理解.隨着時間流逝,系統會按恆定1/QPS時間間隔(如果QPS=100,則間隔是10ms)往桶里加入Token(想象和漏洞漏水相反,有個水龍頭在不斷的加水),如果桶已經滿了就不再加了.新請求來臨時,會各自拿走一個Token,如果沒有Token可拿了就阻塞或者拒絕服務.

令牌桶的另外一個好處是可以方便的改變速度. 一旦需要提高速率,則按需提高放入桶中的令牌的速率. 一般會定時(比如100毫秒)往桶中增加一定數量的令牌, 有些變種算法則實時的計算應該增加的令牌的數量.

3.RateLimiter簡介

Google開源工具包Guava提供了限流工具類RateLimiter,該類基於令牌桶算法(Token Bucket)來完成限流,非常易於使用.RateLimiter經常用於限制對一些物理資源或者邏輯資源的訪問速率.它支持兩種獲取permits接口,一種是如果拿不到立刻返回false,一種會阻塞等待一段時間看能不能拿到.

RateLimiter和Java中的信號量(java.util.concurrent.Semaphore)類似,Semaphore通常用於限制並發量.

源碼注釋中的一個例子,比如我們有很多任務需要執行,但是我們不希望每秒超過兩個任務執行,那么我們就可以使用RateLimiter:

1 2 3 4 5 6 7

final RateLimiter rateLimiter = RateLimiter.create(2.0);   void submitTasks(List<Runnable> tasks, Executor executor) {   for (Runnable task : tasks) {   rateLimiter.acquire(); // may wait   executor.execute(task);   }   }

另外一個例子,假如我們會產生一個數據流,然后我們想以每秒5kb的速度發送出去.我們可以每獲取一個令牌(permit)就發送一個byte的數據,這樣我們就可以通過一個每秒5000個令牌的RateLimiter來實現:

1 2 3 4 5

final RateLimiter rateLimiter = RateLimiter.create(5000.0);   void submitPacket(byte[] packet) {   rateLimiter.acquire(packet.length);   networkService.send(packet);   }

另外,我們也可以使用非阻塞的形式達到降級運行的目的,即使用非阻塞的tryAcquire()方法:

1 2 3 4 5

if(limiter.tryAcquire()) { //未請求到limiter則立即返回false   doSomething();   }else{   doSomethingElse();   }

4.RateLimiter主要接口

RateLimiter其實是一個abstract類,但是它提供了幾個static方法用於創建RateLimiter:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

/**   * 創建一個穩定輸出令牌的RateLimiter，保證了平均每秒不超過permitsPerSecond個請求   * 當請求到來的速度超過了permitsPerSecond，保證每秒只處理permitsPerSecond個請求   * 當這個RateLimiter使用不足(即請求到來速度小於permitsPerSecond)，會囤積最多permitsPerSecond個請求   */   public static RateLimiter create(double permitsPerSecond);       /**   * 創建一個穩定輸出令牌的RateLimiter，保證了平均每秒不超過permitsPerSecond個請求   * 還包含一個熱身期(warmup period),熱身期內，RateLimiter會平滑的將其釋放令牌的速率加大，直到起達到最大速率   * 同樣，如果RateLimiter在熱身期沒有足夠的請求(unused),則起速率會逐漸降低到冷卻狀態   *   * 設計這個的意圖是為了滿足那種資源提供方需要熱身時間，而不是每次訪問都能提供穩定速率的服務的情況(比如帶緩存服務，需要定期刷新緩存的)   * 參數warmupPeriod和unit決定了其從冷卻狀態到達最大速率的時間   */   public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit);

提供了兩個獲取令牌的方法,不帶參數表示獲取一個令牌.如果沒有令牌則一直等待,返回等待的時間(單位為秒),沒有被限流則直接返回0.0:

1 2 3

public double acquire();       public double acquire(int permits);

嘗試獲取令牌,分為待超時時間和不帶超時時間兩種:

1 2 3 4 5 6

public boolean tryAcquire();   //嘗試獲取一個令牌,立即返回   public boolean tryAcquire(int permits);   public boolean tryAcquire(long timeout, TimeUnit unit);   //嘗試獲取permits個令牌,帶超時時間   public boolean tryAcquire(int permits, long timeout, TimeUnit unit);

5.RateLimiter設計

考慮一下RateLimiter是如何設計的,並且為什么要這樣設計.

RateLimiter的主要功能就是提供一個穩定的速率,實現方式就是通過限制請求流入的速度,比如計算請求等待合適的時間閾值.

實現QPS速率的最簡單的方式就是記住上一次請求的最后授權時間,然后保證1/QPS秒內不允許請求進入.比如QPS=5,如果我們保證最后一個被授權請求之后的200ms的時間內沒有請求被授權,那么我們就達到了預期的速率.如果一個請求現在過來但是最后一個被授權請求是在100ms之前,那么我們就要求當前這個請求等待100ms.按照這個思路,請求15個新令牌(許可證)就需要3秒.

有一點很重要:上面這個設計思路的RateLimiter記憶非常的淺,它的腦容量非常的小,只記得上一次被授權的請求的時間.如果RateLimiter的一個被授權請求q之前很長一段時間沒有被使用會怎么樣?這個RateLimiter會立馬忘記過去這一段時間的利用不足,而只記得剛剛的請求q.

過去一段時間的利用不足意味着有過剩的資源是可以利用的.這種情況下,RateLimiter應該加把勁(speed up for a while)將這些過剩的資源利用起來.比如在向網絡中發生數據的場景(限流),過去一段時間的利用不足可能意味着網卡緩沖區是空的,這種場景下,我們是可以加速發送來將這些過程的資源利用起來.

另一方面,過去一段時間的利用不足可能意味着處理請求的服務器對即將到來的請求是准備不足的(less ready for future requests),比如因為很長一段時間沒有請求當前服務器的cache是陳舊的,進而導致即將到來的請求會觸發一個昂貴的操作(比如重新刷新全量的緩存).

為了處理這種情況,RateLimiter中增加了一個維度的信息,就是過去一段時間的利用不足(past underutilization),代碼中使用storedPermits變量表示.當沒有利用不足這個變量為0,最大能達到maxStoredPermits(maxStoredPermits表示完全沒有利用).因此,請求的令牌可能從兩個地方來:

1.  
   1.過去剩余的令牌(stored permits, 可能沒有)
2.  
   2.現有的令牌(fresh permits,當前這段時間還沒用完的令牌)

我們將通過一個例子來解釋它是如何工作的:

對一個每秒產生一個令牌的RateLimiter,每有一個沒有使用令牌的一秒,我們就將storedPermits加1,如果RateLimiter在10秒都沒有使用,則storedPermits變成10.0.這個時候,一個請求到來並請求三個令牌(acquire(3)),我們將從storedPermits中的令牌為其服務,storedPermits變為7.0.這個請求之后立馬又有一個請求到來並請求10個令牌,我們將從storedPermits剩余的7個令牌給這個請求,剩下還需要三個令牌,我們將從RateLimiter新產生的令牌中獲取.我們已經知道,RateLimiter每秒新產生1個令牌,就是說上面這個請求還需要的3個請求就要求其等待3秒.

想象一個RateLimiter每秒產生一個令牌,現在完全沒有使用(處於初始狀態),限制一個昂貴的請求acquire(100)過來.如果我們選擇讓這個請求等待100秒再允許其執行,這顯然很荒謬.我們為什么什么也不做而只是傻傻的等待100秒,一個更好的做法是允許這個請求立即執行(和acquire(1)沒有區別),然后將隨后到來的請求推遲到正確的時間點.這種策略,我們允許這個昂貴的任務立即執行,並將隨后到來的請求推遲100秒.這種策略就是讓任務的執行和等待同時進行.

一個重要的結論:RateLimiter不會記最后一個請求,而是即下一個請求允許執行的時間.這也可以很直白的告訴我們到達下一個調度時間點的時間間隔.然后定一個一段時間未使用的Ratelimiter也很簡單:下一個調度時間點已經過去,這個時間點和現在時間的差就是Ratelimiter多久沒有被使用,我們會將這一段時間翻譯成storedPermits.所有,如果每秒鍾產生一個令牌(rate==1),並且正好每秒來一個請求,那么storedPermits就不會增長.

6.RateLimiter主要源碼

RateLimiter定義了兩個create函數用於構建不同形式的RateLimiter:

1.  
   1.public static RateLimiter create(double permitsPerSecond)
2.  
   用於創建SmoothBursty類型的RateLimiter
3.  
   2.public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit)
4.  
   用於創建
5.  
    

源碼下面以acquire為例子,分析一下RateLimiter如何實現限流:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

public double acquire() {   return acquire(1);   }   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);   }

下面方法來自RateLimiter的具體實現類SmoothRateLimiter:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

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 = nextFreeTicketMicros + waitMicros;   this.storedPermits -= storedPermitsToSpend;   return returnValue;   }   private void resync(long nowMicros) {   // if nextFreeTicket is in the past, resync to now   if (nowMicros > nextFreeTicketMicros) {   storedPermits = min(maxPermits,   storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);   nextFreeTicketMicros = nowMicros;   }   }

另外，對於storedPermits的使用，RateLimiter存在兩種策略，二者區別主要體現在使用storedPermits時候需要等待的時間。這個邏輯由storedPermitsToWaitTime函數實現：

1 2 3 4 5 6 7 8 9

/**   * 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);

存在兩種策略就是為了應對我們上面講到的，存在資源使用不足大致分為兩種情況： (1).資源確實使用不足，這些剩余的資源我們私海可以使用的； (2).提供資源的服務過去還沒准備好，比如服務剛啟動等；

為此，RateLimiter實際上由兩種實現策略，其實現分別見SmoothBursty和SmoothWarmingUp。二者主要的區別就是storedPermitsToWaitTime實現以及maxPermits數量的計算。

6.1 SmoothBursty

SmoothBursty使用storedPermits不需要額外等待時間。並且默認maxBurstSeconds未1，因此maxPermits為permitsPerSecond，即最多可以存儲1秒的剩余令牌，比如QPS=5，則maxPermits=5.

下面這個RateLimiter的入口就是用來創建SmoothBursty類型的RateLimiter，

1	public static RateLimiter create(double permitsPerSecond)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33

/**   * This implements a "bursty" RateLimiter, where storedPermits are translated to   * zero throttling. The maximum number of permits that can be saved (when the RateLimiter is   * unused) is defined in terms of time, in this sense: if a RateLimiter is 2qps, and this   * time is specified as 10 seconds, we can save up to 2 * 10 = 20 permits.   */   static final class SmoothBursty extends SmoothRateLimiter {   /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */   final double maxBurstSeconds;       SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {   super(stopwatch);   this.maxBurstSeconds = maxBurstSeconds;   }       void doSetRate(double permitsPerSecond, double stableIntervalMicros) {   double oldMaxPermits = this.maxPermits;   maxPermits = maxBurstSeconds * permitsPerSecond;   System.out.println("maxPermits=" + maxPermits);   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;   }   }       long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {   return 0L;   }   }

一個簡單的使用示意圖及解釋，下面私海一個QPS=4的SmoothBursty:

1.  
   (1).t=0,這時候storedPermits=0，請求1個令牌，等待時間=0；
2.  
   (2).t=1,這時候storedPermits=3，請求3個令牌，等待時間=0；
3.  
   (3).t=2,這時候storedPermits=4，請求10個令牌，等待時間=0，超前使用了2個令牌；
4.  
   (4).t=3,這時候storedPermits=0，請求1個令牌，等待時間=0.5；

代碼的輸出：

1 2 3 4 5 6 7 8 9 10 11

maxPermits=4.0, storedPermits=7.2E-4, stableIntervalMicros=250000.0, nextFreeTicketMicros=1472   acquire(1), sleepSecond=0.0       maxPermits=4.0, storedPermits=3.012212, stableIntervalMicros=250000.0, nextFreeTicketMicros=1004345   acquire(3), sleepSecond=0.0       maxPermits=4.0, storedPermits=4.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=2004668   acquire(10), sleepSecond=0.0       maxPermits=4.0, storedPermits=0.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=3504668   acquire(1), sleepSecond=0.499591

6.2 SmoothWarmingUp

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52

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 halfPermits;       SmoothWarmingUp(SleepingStopwatch stopwatch, long warmupPeriod, TimeUnit timeUnit) {   super(stopwatch);   this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod);   }       @Override   void doSetRate(double permitsPerSecond, double stableIntervalMicros) {   double oldMaxPermits = maxPermits;   maxPermits = warmupPeriodMicros / stableIntervalMicros;   halfPermits = maxPermits / 2.0;   // Stable interval is x, cold is 3x, so on average it's 2x. Double the time -> halve the rate   double coldIntervalMicros = stableIntervalMicros * 3.0;   slope = (coldIntervalMicros - stableIntervalMicros) / halfPermits;   if (oldMaxPermits == Double.POSITIVE_INFINITY) {   // if we don't special-case this, we would get storedPermits == NaN, below   storedPermits = 0.0;   } else {   storedPermits = (oldMaxPermits == 0.0)   ? maxPermits // initial state is cold   : storedPermits * maxPermits / oldMaxPermits;   }   }       @Override   long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {   double availablePermitsAboveHalf = storedPermits - halfPermits;   long micros = 0;   // measuring the integral on the right part of the function (the climbing line)   if (availablePermitsAboveHalf > 0.0) {   double permitsAboveHalfToTake = min(availablePermitsAboveHalf, permitsToTake);   micros = (long) (permitsAboveHalfToTake * (permitsToTime(availablePermitsAboveHalf)   + permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0);   permitsToTake -= permitsAboveHalfToTake;   }   // measuring the integral on the left part of the function (the horizontal line)   micros += (stableIntervalMicros * permitsToTake);   return micros;   }       private double permitsToTime(double permits) {   return stableIntervalMicros + permits * slope;   }   }

maxPermits等於熱身(warmup)期間能產生的令牌數，比如QPS=4，warmup為2秒，則maxPermits=8.halfPermits為maxPermits的一半.

參考注釋中的神圖：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

* ^ throttling   * |   * 3*stable + /   * interval | /.   * (cold) | / .   * | / . <-- "warmup period" is the area of the trapezoid between   * 2*stable + / . halfPermits and maxPermits   * interval | / .   * | / .   * | / .   * stable +----------/ WARM . }   * interval | . UP . } <-- this rectangle (from 0 to maxPermits, and   * | . PERIOD. } height == stableInterval) defines the cooldown period,   * | . . } and we want cooldownPeriod == warmupPeriod   * |---------------------------------> storedPermits   * (halfPermits) (maxPermits)   *

下面是我們QPS=4，warmup為2秒時候對應的圖。 

maxPermits=8，halfPermits=4，和SmoothBursty相同的請求序列：

1.  
   (1).t=0,這時候storedPermits=8，請求1個令牌，使用1個storedPermits消耗時間=1×(0.75+0.625)/2=0.6875秒；
2.  
   (2).t=1,這時候storedPermits=8，請求3個令牌，使用3個storedPermits消耗時間=3×(0.75+0.375)/2=1.6875秒(注意已經超過1秒了，意味着下次產生新Permit時間為2.6875)；
3.  
   (3).t=2,這時候storedPermits=5，請求10個令牌，使用5個storedPermits消耗時間=1×(0.375+0.25)/2+4*0.25=1.3125秒，再加上額外請求的5個新產生的Permit需要消耗=5*0.25=1.25秒，即總共需要耗時2.5625秒，則下一次產生新的Permit時間為2.6875+2.5625=5.25，注意當前請求私海2.6875才返回的，之前一直阻塞；
4.  
   (4).t=3,因為前一個請求阻塞到2.6875，實際這個請求3.6875才到達RateLimiter，請求1個令牌，storedPermits=0，下一次產生新Permit時間為5.25，因此總共需要等待5.25-3.6875=1.5625秒；

實際執行結果：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

warmupPeriodMicros=2000000   stableIntervalMicros=250000.0, maxPermits=8.0, halfPermits=4.0, coldIntervalMicros=750000.0, slope=125000.0, storedPermits=8.0       maxPermits=8.0, storedPermits=8.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=1524   acquire(1), sleepSecond=0.0       maxPermits=8.0, storedPermits=8.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=1001946   acquire(3), sleepSecond=0.0       maxPermits=8.0, storedPermits=5.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=2689446   acquire(10), sleepSecond=0.687186       maxPermits=8.0, storedPermits=0.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=5251946   acquire(1), sleepSecond=1.55