轉載: https://cloud.tencent.com/developer/article/1408819
限流是保護高並發系統的三把利器之一，另外兩個是緩存和降級。限流在很多場景中用來限制並發和請求量，比如說秒殺搶購，保護自身系統和下游系統不被巨型流量沖垮等。
限流的目的是通過對並發訪問/請求進行限速或者一個時間窗口內的的請求進行限速來保護系統，一旦達到限制速率則可以拒絕服務或進行流量整形。
常用的限流方式和場景有：限制總並發數（比如數據庫連接池、線程池）、限制瞬時並發數（如nginx的limit_conn模塊，用來限制瞬時並發連接數，Java的Semaphore也可以實現）、限制時間窗口內的平均速率（如Guava的RateLimiter、nginx的limit_req模塊，限制每秒的平均速率）；其他還有如限制遠程接口調用速率、限制MQ的消費速率。另外還可以根據網絡連接數、網絡流量、CPU或內存負載等來限流。
比如說，我們需要限制方法被調用的並發數不能超過100（同一時間並發數），則我們可以用信號量Semaphore實現。可如果我們要限制方法在一段時間內平均被調用次數不超過100，則需要使用RateLimiter。
限流的基礎算法
我們先來講解一下兩個限流相關的基本算法：漏桶算法和令牌桶算法。
漏桶算法
從上圖中，我們可以看到，就像一個漏斗一樣，進來的水量就好像訪問流量一樣，而出去的水量就像是我們的系統處理請求一樣。當訪問流量過大時，這個漏斗中就會積水，如果水太多了就會溢出。
漏桶算法的實現往往依賴於隊列，請求到達如果隊列未滿則直接放入隊列，然后有一個處理器按照固定頻率從隊列頭取出請求進行處理。如果請求量大，則會導致隊列滿，那么新來的請求就會被拋棄。
漏桶算法示意圖
令牌桶算法則是一個存放固定容量令牌的桶，按照固定速率往桶里添加令牌。桶中存放的令牌數有最大上限，超出之后就被丟棄或者拒絕。當流量或者網絡請求到達時，每個請求都要獲取一個令牌，如果能夠獲取到，則直接處理，並且令牌桶刪除一個令牌。如果獲取不同，該請求就要被限流，要么直接丟棄，要么在緩沖區等待。
令牌桶算法示意圖
令牌桶和漏桶對比：
- 令牌桶是按照固定速率往桶中添加令牌，請求是否被處理需要看桶中令牌是否足夠，當令牌數減為零時則拒絕新的請求；漏桶則是按照常量固定速率流出請求，流入請求速率任意，當流入的請求數累積到漏桶容量時，則新流入的請求被拒絕；
- 令牌桶限制的是平均流入速率，允許突發請求，只要有令牌就可以處理，支持一次拿3個令牌，4個令牌；漏桶限制的是常量流出速率，即流出速率是一個固定常量值，比如都是1的速率流出，而不能一次是1，下次又是2，從而平滑突發流入速率；
- 令牌桶允許一定程度的突發，而漏桶主要目的是平滑流出速率；
Guava RateLimiter
Guava是Java領域優秀的開源項目，它包含了Google在Java項目中使用一些核心庫，包含集合(Collections)，緩存(Caching)，並發編程庫(Concurrency)，常用注解(Common annotations)，String操作，I/O操作方面的眾多非常實用的函數。
Guava的RateLimiter提供了令牌桶算法實現：平滑突發限流(SmoothBursty)和平滑預熱限流(SmoothWarmingUp)實現。
類圖
RateLimiter的類圖如上所示，其中RateLimiter是入口類，它提供了兩套工廠方法來創建出兩個子類。這很符合《Effective Java》中的用靜態工廠方法代替構造函數的建議，畢竟該書的作者也正是Guava庫的主要維護者，二者配合"食用"更佳。
// RateLimiter提供了兩個工廠方法，最終會調用下面兩個函數，生成RateLimiter的兩個子類。
static RateLimiter create(SleepingStopwatch stopwatch, double permitsPerSecond) { RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */); rateLimiter.setRate(permitsPerSecond); return rateLimiter; } static RateLimiter create( SleepingStopwatch stopwatch, double permitsPerSecond, long warmupPeriod, TimeUnit unit, double coldFactor) { RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor); rateLimiter.setRate(permitsPerSecond); return rateLimiter; }
平滑突發限流
使用RateLimiter的靜態方法創建一個限流器，設置每秒放置的令牌數為5個。返回的RateLimiter對象可以保證1秒內不會給超過5個令牌，並且以固定速率進行放置，達到平滑輸出的效果。
public void testSmoothBursty() { RateLimiter r = RateLimiter.create(5); while (true) { System.out.println("get 1 tokens: " + r.acquire() + "s"); } /** * output: 基本上都是0.2s執行一次，符合一秒發放5個令牌的設定。 * get 1 tokens: 0.0s * get 1 tokens: 0.182014s * get 1 tokens: 0.188464s * get 1 tokens: 0.198072s * get 1 tokens: 0.196048s * get 1 tokens: 0.197538s * get 1 tokens: 0.196049s */ }
RateLimiter使用令牌桶算法，會進行令牌的累積，如果獲取令牌的頻率比較低，則不會導致等待，直接獲取令牌。
public void testSmoothBursty2() { RateLimiter r = RateLimiter.create(2); while (true) { System.out.println("get 1 tokens: " + r.acquire(1) + "s"); try { Thread.sleep(2000); } catch (Exception e) {} System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("end"); /** * output: * get 1 tokens: 0.0s * get 1 tokens: 0.0s * get 1 tokens: 0.0s * get 1 tokens: 0.0s * end * get 1 tokens: 0.499796s * get 1 tokens: 0.0s * get 1 tokens: 0.0s * get 1 tokens: 0.0s */ } }
RateLimiter由於會累積令牌，所以可以應對突發流量。在下面代碼中，有一個請求會直接請求5個令牌，但是由於此時令牌桶中有累積的令牌，足以快速響應。
RateLimiter在沒有足夠令牌發放時，采用滯后處理的方式，也就是前一個請求獲取令牌所需等待的時間由下一次請求來承受，也就是代替前一個請求進行等待。
public void testSmoothBursty3() { RateLimiter r = RateLimiter.create(5); while (true) { System.out.println("get 5 tokens: " + r.acquire(5) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("end"); /** * output: * get 5 tokens: 0.0s * get 1 tokens: 0.996766s 滯后效應，需要替前一個請求進行等待 * get 1 tokens: 0.194007s * get 1 tokens: 0.196267s * end * get 5 tokens: 0.195756s * get 1 tokens: 0.995625s 滯后效應，需要替前一個請求進行等待 * get 1 tokens: 0.194603s * get 1 tokens: 0.196866s */ } }
平滑預熱限流
RateLimiter的SmoothWarmingUp是帶有預熱期的平滑限流，它啟動后會有一段預熱期，逐步將分發頻率提升到配置的速率。
比如下面代碼中的例子，創建一個平均分發令牌速率為2，預熱期為3分鍾。由於設置了預熱時間是3秒，令牌桶一開始並不會0.5秒發一個令牌，而是形成一個平滑線性下降的坡度，頻率越來越高，在3秒鍾之內達到原本設置的頻率，以后就以固定的頻率輸出。這種功能適合系統剛啟動需要一點時間來“熱身”的場景。
public void testSmoothwarmingUp() { RateLimiter r = RateLimiter.create(2, 3, TimeUnit.SECONDS); while (true) { System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("get 1 tokens: " + r.acquire(1) + "s"); System.out.println("end"); /** * output: * get 1 tokens: 0.0s * get 1 tokens: 1.329289s * get 1 tokens: 0.994375s * get 1 tokens: 0.662888s 上邊三次獲取的時間相加正好為3秒 * end * get 1 tokens: 0.49764s 正常速率0.5秒一個令牌 * get 1 tokens: 0.497828s * get 1 tokens: 0.49449s * get 1 tokens: 0.497522s */ } }
源碼分析
看完了RateLimiter的基本使用示例后，我們來學習一下它的實現原理。先了解一下幾個比較重要的成員變量的含義。
//SmoothRateLimiter.java
//當前存儲令牌數 double storedPermits; //最大存儲令牌數 double maxPermits; //添加令牌時間間隔 double stableIntervalMicros; /** * 下一次請求可以獲取令牌的起始時間 * 由於RateLimiter允許預消費，上次請求預消費令牌后 * 下次請求需要等待相應的時間到nextFreeTicketMicros時刻才可以獲取令牌 */ private long nextFreeTicketMicros = 0L;
平滑突發限流
RateLimiter的原理就是每次調用acquire時用當前時間和nextFreeTicketMicros進行比較，根據二者的間隔和添加單位令牌的時間間隔stableIntervalMicros來刷新存儲令牌數storedPermits。然后acquire會進行休眠，直到nextFreeTicketMicros。
acquire函數如下所示，它會調用reserve函數計算獲取目標令牌數所需等待的時間，然后使用SleepStopwatch進行休眠，最后返回等待時間。
public double acquire(int permits) { // 計算獲取令牌所需等待的時間 long microsToWait = reserve(permits); // 進行線程sleep stopwatch.sleepMicrosUninterruptibly(microsToWait); return 1.0 * microsToWait / SECONDS.toMicros(1L); } final long reserve(int permits) { checkPermits(permits); // 由於涉及並發操作，所以使用synchronized進行並發操作 synchronized (mutex()) { return reserveAndGetWaitLength(permits, stopwatch.readMicros()); } } final long reserveAndGetWaitLength(int permits, long nowMicros) { // 計算從當前時間開始，能夠獲取到目標數量令牌時的時間 long momentAvailable = reserveEarliestAvailable(permits, nowMicros); // 兩個時間相減，獲得需要等待的時間 return max(momentAvailable - nowMicros, 0); }
reserveEarliestAvailable是刷新令牌數和下次獲取令牌時間nextFreeTicketMicros的關鍵函數。它有三個步驟，一是調用resync函數增加令牌數，二是計算預支付令牌所需額外等待的時間，三是更新下次獲取令牌時間nextFreeTicketMicros和存儲令牌數storedPermits。
這里涉及RateLimiter的一個特性，也就是可以預先支付令牌，並且所需等待的時間在下次獲取令牌時再實際執行。詳細的代碼邏輯的解釋請看注釋。
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) { // 刷新令牌數，相當於每次acquire時在根據時間進行令牌的刷新 resync(nowMicros); long returnValue = nextFreeTicketMicros; // 獲取當前已有的令牌數和需要獲取的目標令牌數進行比較，計算出可以目前即可得到的令牌數。 double storedPermitsToSpend = min(requiredPermits, this.storedPermits); // freshPermits是需要預先支付的令牌，也就是目標令牌數減去目前即可得到的令牌數 double freshPermits = requiredPermits - storedPermitsToSpend; // 因為會突然涌入大量請求，而現有令牌數又不夠用，因此會預先支付一定的令牌數 // waitMicros即是產生預先支付令牌的數量時間，則將下次要添加令牌的時間應該計算時間加上watiMicros long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) + (long) (freshPermits * stableIntervalMicros); // storedPermitsToWaitTime在SmoothWarmingUp和SmoothBuresty的實現不同，用於實現預熱緩沖期 // SmoothBuresty的storedPermitsToWaitTime直接返回0，所以watiMicros就是預先支付的令牌所需等待的時間 try { // 更新nextFreeTicketMicros,本次預先支付的令牌所需等待的時間讓下一次請求來實際等待。 this.nextFreeTicketMicros = LongMath.checkedAdd(nextFreeTicketMicros, waitMicros); } catch (ArithmeticException e) { this.nextFreeTicketMicros = Long.MAX_VALUE; } // 更新令牌數，最低數量為0 this.storedPermits -= storedPermitsToSpend; // 返回舊的nextFreeTicketMicros數值，無需為預支付的令牌多加等待時間。 return returnValue; } // SmoothBurest long storedPermitsToWaitTime(double storedPermits, double permitsToTake) { return 0L; }
resync函數用於增加存儲令牌，核心邏輯就是(nowMicros - nextFreeTicketMicros) / stableIntervalMicros。當前時間大於nextFreeTicketMicros時進行刷新，否則直接返回。
void resync(long nowMicros) { // 當前時間晚於nextFreeTicketMicros，所以刷新令牌和nextFreeTicketMicros if (nowMicros > nextFreeTicketMicros) { // coolDownIntervalMicros函數獲取每機秒生成一個令牌，SmoothWarmingUp和SmoothBuresty的實現不同 // SmoothBuresty的coolDownIntervalMicros直接返回stableIntervalMicros // 當前時間減去要更新令牌的時間獲取時間間隔，再除以添加令牌時間間隔獲取這段時間內要添加的令牌數 storedPermits = min(maxPermits, storedPermits + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros()); nextFreeTicketMicros = nowMicros; } // 如果當前時間早於nextFreeTicketMicros，則獲取令牌的線程要一直等待到nextFreeTicketMicros,該線程獲取令牌所需 // 額外等待的時間由下一次獲取的線程來代替等待。 } double coolDownIntervalMicros() { return stableIntervalMicros; }
下面我們舉個例子，讓大家更好的理解resync和reserveEarliestAvailable函數的邏輯。
比如RateLimiter的stableIntervalMicros為500，也就是1秒發兩個令牌，storedPermits為0，nextFreeTicketMicros為1553918495748。線程一acquire(2)，當前時間為1553918496248，首先resync函數計算，(1553918496248 - 1553918495748)/500 = 1，所以當前可獲取令牌數為1，但是由於可以預支付，所以nextFreeTicketMicros= nextFreeTicketMicro + 1 * 500 = 1553918496748。線程一無需等待。
緊接着，線程二也來acquire(2)，首先resync函數發現當前時間早於nextFreeTicketMicros，所以無法增加令牌數，所以需要預支付2個令牌，nextFreeTicketMicros= nextFreeTicketMicro + 2 * 500 = 1553918497748。線程二需要等待1553918496748時刻，也就是線程一獲取時計算的nextFreeTicketMicros時刻。同樣的，線程三獲取令牌時也需要等待到線程二計算的nextFreeTicketMicros時刻。
平滑預熱限流
上述就是平滑突發限流RateLimiter的實現，下面我們來看一下加上預熱緩沖期的實現原理。
SmoothWarmingUp實現預熱緩沖的關鍵在於其分發令牌的速率會隨時間和令牌數而改變，速率會先慢后快。表現形式如下圖所示，令牌刷新的時間間隔由長逐漸變短。等存儲令牌數從maxPermits到達thresholdPermits時，發放令牌的時間價格也由coldInterval降低到了正常的stableInterval。
image.png
SmoothWarmingUp的相關代碼如下所示，相關的邏輯都寫在注釋中。
// SmoothWarmingUp，等待時間就是計算上圖中梯形或者正方形的面積。
long storedPermitsToWaitTime(double storedPermits, double permitsToTake) { /** * 當前permits超出閾值的部分 */ double availablePermitsAboveThreshold = storedPermits - thresholdPermits; long micros = 0; /** * 如果當前存儲的令牌數超出thresholdPermits */ if (availablePermitsAboveThreshold > 0.0) { /** * 在閾值右側並且需要被消耗的令牌數量 */ double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake); /** * 梯形的面積 * * 高 * (頂 * 底) / 2 * * 高是 permitsAboveThresholdToTake 也就是右側需要消費的令牌數 * 底 較長 permitsToTime(availablePermitsAboveThreshold) * 頂 較短 permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake) */ micros = (long) (permitsAboveThresholdToTake * (permitsToTime(availablePermitsAboveThreshold) + permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake)) / 2.0); /** * 減去已經獲取的在閾值右側的令牌數 */ permitsToTake -= permitsAboveThresholdToTake; } /** * 平穩時期的面積，正好是長乘寬 */ micros += (stableIntervalMicros * permitsToTake); return micros; } double coolDownIntervalMicros() { /** * 每秒增加的令牌數為 warmup時間/maxPermits. 這樣的話，在warmuptime時間內，就就增張的令牌數量 * 為 maxPermits */ return warmupPeriodMicros / maxPermits; }
后記
RateLimiter只能用於單機的限流，如果想要集群限流，則需要引入redis或者阿里開源的sentinel中間件，請大家繼續關注。