限流原理解讀之guava中的RateLimiter
RateLimiter有兩種新建的方式
-
建立Bursty方式
-
建立WarmingUp方式
以下原始碼來自 guava-17.0
Bursty
//初始化 RateLimiter r = RateLimiter.create(1); //不阻塞 r.tryAcquire(); //阻塞 r.acquire() 複製程式碼
RateLimiter.create做了兩件事情建立Bursty物件和設定了速率,至次初始化過程結束
RateLimiter rateLimiter = new Bursty(ticker, 1.0 /* maxBurstSeconds */); //ticker預設使用自己定義的 rateLimiter.setRate(permitsPerSecond); 複製程式碼
- 新建Bursty物件。它指定的是能夠儲存的最大時間是多長，比如設定的時間是1s,那麼假設允許每秒鐘發放的令牌數量為2，能儲存的最大量為2；
-
setRate。 內部通過私有鎖來保證速率的修改是執行緒安全的
synchronized (mutex) { //1：檢視當前的時間是否比預計下次可發放令牌的時間要大，如果大，更新下次可發放令牌的時間為當前時間 resync(readSafeMicros()); //2：計算兩次發放令牌之間的時間間隔，比如1s中需要發放5個，那它就是 200000.0微秒 double stableIntervalMicros = TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond; this.stableIntervalMicros = stableIntervalMicros; //3：設定maxPermits和storedPermits doSetRate(permitsPerSecond, stableIntervalMicros); } 複製程式碼
resync原始碼
private void resync(long nowMicros) {
// 檢視當前的時間是否比預計下次可發放令牌的時間要大，如果大，更新下次可發放令牌的時間為當前時間
if (nowMicros > nextFreeTicketMicros) {
storedPermits = Math.min(maxPermits,
storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);
nextFreeTicketMicros = nowMicros;
}
}
複製程式碼
doSetRate原始碼
@Overridevoid doSetRate(double permitsPerSecond, double stableIntervalMicros) {
double oldMaxPermits = this.maxPermits;
maxPermits = maxBurstSeconds * permitsPerSecond;
storedPermits = (oldMaxPermits == 0.0)
? 0.0 // 初始條件儲存的是沒有
storedPermits * maxPermits / oldMaxPermits;
}
複製程式碼
在整個的初始化過程中，關鍵資訊是：
-
nextFreeTicketMicros 預計下次發放令牌的時間, stableIntervalMicros 兩次發放令牌之間的時間間隔
-
maxPermits 最大能儲存的令牌的數量 storedPermits 已經儲存的令牌數
為什麼是nextFreeTicketMicros?
最簡單的維持QPS速率的方式就是記住最後一次請求的時間，然後確保再次有請求過來的時候，已經經過了 1/QPS 秒。比如QPS是5 次/秒，只需要確保兩次請求時間經過了200ms即可，如果剛好在100ms到達，就會再等待100ms,也就是說，如果一次性需要15個令牌，需要的時間為為3s。但是對於一個長時間沒有請求的系統，這樣的的設計方式有一定的不合理之處。考慮一個場景：如果一個RateLimiter,每秒產生1個令牌,它一直沒有使用過，突然來了一個需要100個令牌的請求，選擇等待100s再執行這個請求，顯得不太明智，更好的處理方式為立即執行它，然後把接下來的請求推遲100s。
因而RateLimiter本身並不記下最後一次請求的時間，而是記下下一次期望執行的時間（nextFreeTicketMicros）。
這種方式帶來的一個好處是，可以去判斷等待的超時時間是否大於下次執行的時間，以使得能夠執行，如果等待的超時時間太短，就能立即返回。
為什麼會有一個標記代表儲存了多少令牌？
同樣的考慮長時間沒有使用的場景。如果長時間沒有請求，突然間來了，這個時候是否應該立馬放行這些請求？長時間沒有使用可能意味著兩件事：
- 很多資源是存在空閒的情況，比如說網路請求長時間沒有，它的緩衝區很有可能是空的，此時是可以加速傳輸，提高它的利用率
- 一些時候，瞬間的爆發會導致溢位，比如說服務上的快取過期了，需要去查詢庫，這個花銷是非常“昂貴”的，過多的請求會導致資料庫撐不住
RateLimiter就使用storedPermits來給過去請求的不充分程度建模。它的儲存規則如下： 假設RateLimiter每秒產生一個令牌,每過去一秒如果沒有請求，RateLimter也就沒有消費，就使storedPermits增長1。假設10s之內都沒有請求過來,storedPermits就變成了10（假設maxPermits>10），此時如果要獲取3個令牌，會使用storedPermits來中的令牌來處理，然後它的值變為了7，片刻之後，如果呼叫了acquire(10),部分的會從storedPermits拿到7個許可權，剩餘的3個則需要重新產生。
總的來說RateLimiter提供了一個storedPermits變數，當資源利用充分的時候，它就是0，最大可以增長到 maxStoredPermits。請求所需的令牌來自於兩個地方：stored permits(空閒時儲存的令牌)和fresh permits（現有的令牌）
怎麼衡量從storedPermits中獲取令牌這個過程？
同樣假設每秒RateLimiter只生產一個令牌，正常情況下，如果一次來了3個請求，整個過程會持續3秒鐘。考慮到長時間沒有請求的場景：
- 資源空閒。這種時候系統是能承受住一定量的請求的，當然希望在承受範圍之內能夠更快的提供請求，也就是說，如果有儲存令牌，相比新產生令牌，此時希望能夠更快的獲取令牌，也就是此時從儲存令牌中獲取令牌的時間消耗要比產生新令牌要少，從而更快相應請求
- 瞬時流量過大。這時候就不希望過快的消耗儲存的令牌，希望它能夠相比產生新的令牌的時間消耗大些，從而能夠使請求相對平緩。
分析可知，針對不同的場景，需要對獲取storedPermits做不同的處理，Ratelimiter的實現方式就是 storedPermitsToWaitTime 函式，它建立了從storedPermits中獲取令牌和時間花銷的模型函式,而衡量時間的花銷就是通過對模型函式進行積分計算，比如原來儲存了10個令牌，現在需要拿3個令牌，還剩餘7個，那麼所需要的時間花銷就是該函式從7-10區間中的積分。
這種方式保證了任何獲取令牌方式所需要的時間都是一樣的，好比 每次拿一個和先拿兩個再拿一個，從時間上來講並沒有分別。
storedPermitsToWaitTime實現原理
storedPermits本身是用來衡量沒有使用的時間的。在沒有使用令牌的時候儲存，儲存的速率（單位時間記憶體儲的令牌的個數）是 每沒用1次就儲存1次: rate=permites/time 。也就是說 1 / rate = time / permits，那麼可得到 (1/rate)*permits 就可以來衡量時間花銷。
選取(1/rate)作為基準線
- 如果選取一條在它之上的線，就做到了比從fresh permits中獲取要慢；
- 如果在基準線之下，則是比從fresh permits中獲取要快；
- 剛好是基準線，那麼從storedPermits中獲取和新產生的速率一模一樣；
Bursty的storedPermitsToWaitTime函式實現
long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
return 0L;
}
複製程式碼
它直接返回了0，也就是在基準線之下，獲取storedPermits的速率比新產生要快，立即能夠拿到儲存的量
WarmingUp
//初始化 RateLimiter r =RateLimiter.create(1,1,TimeUnit.SECONDS); //不阻塞 r.tryAcquire(); //阻塞 r.acquire() 複製程式碼
create方法建立了WarmingUp物件，並這隻了對應的速率
RateLimiter rateLimiter = new WarmingUp(ticker, warmupPeriod, unit); rateLimiter.setRate(permitsPerSecond); 複製程式碼
相比Bursty，它多了個引數warmupPeroid,它會以提供的unit為時間單位，轉換成微秒儲存。setRate類似於Bursty，只是在doSetRate提供不同的實現
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
double oldMaxPermits = maxPermits;
//1:最大的儲存個數為需要預熱的時間除以兩個請求的時間間隔，比如設定預熱時間為1s,每秒有5個請求，那麼最大的儲存個數為1000ms/200ms=5個
maxPermits = warmupPeriodMicros / stableIntervalMicros;
//2:計算最大儲存permits的一半
halfPermits = maxPermits / 2.0;
//3:初始化穩定時間間隔的3倍作為冷卻時間間隔
double coldIntervalMicros = stableIntervalMicros * 3.0;
//4:設定基準線的斜率
slope = (coldIntervalMicros - stableIntervalMicros) / halfPermits;
if (oldMaxPermits == Double.POSITIVE_INFINITY) {
storedPermits = 0.0;
} else {
storedPermits = (oldMaxPermits == 0.0)
? maxPermits // 初始條件下，認為就是儲存滿的，以達到緩慢消費的效果
: storedPermits * maxPermits / oldMaxPermits;
}
}
複製程式碼
在這個過程中可以看到Warmup方式新增了一個halfPermits的設計，以及通過公式slope=(coldIntervalMicros-stableIntervalMicros)/halfPermits
，他們在函式 storedPermitsToWaitTime中得到了運用
@Overridelong storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
//1:計算儲存的令牌中超過了最大令牌一半的數量
double availablePermitsAboveHalf = storedPermits - halfPermits;
long micros = 0;
// 計算超過一半的部分所需要的時間花銷(對於函式來說，就是積分計算)
if (availablePermitsAboveHalf > 0.0) {
double permitsAboveHalfToTake = Math.min(availablePermitsAboveHalf, permitsToTake);
micros = (long) (permitsAboveHalfToTake * (permitsToTime(availablePermitsAboveHalf)
+ permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0);
permitsToTake -= permitsAboveHalfToTake;
}
// 計算函式的尚未超過一半的部分所需要的時間花銷
micros += (stableIntervalMicros * permitsToTake);
return micros;
}
private double permitsToTime(double permits) {
return stableIntervalMicros + permits * slope;
}
複製程式碼
WarmingUp的設計理念
WarmingUp對時間花銷衡量方式為下圖
*^ 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) 複製程式碼
橫軸表示儲存的令牌個數，縱軸表示時間，這樣函式的積分就可以表示所要消耗的時間。
在程式剛開始執行的時候，warmingup方式會存滿所有的令牌，而根據從儲存令牌中的獲取方式，可以實現從儲存最大令牌中到降到一半令牌所需要的時間為儲存同量令牌時間的2倍，從而使得剛開始的時候發放令牌的速度比較慢，等消耗一半之後，獲取的速率和生產的速率一致，從而也就實現了一個‘熱身’的概念
從storedPermits中獲取令牌所需要的時間，它分為兩部分，以maxPetmits的一半為分割點
-
storedPermits <= halfPermits 的時候,儲存和消費storedPermits的速率與產生的速率一模一樣
-
storedPermits>halfPermits, 儲存storePermites所需要的時間和產生的速率保持一致，但是消費storePermites從maxPermits到halfPermits所需要的時間為從halfPermits增長到maxPermits所需要時間的2被，也就是比新令牌產生要慢 為什麼在分隔點計算還有斜率方面選了3倍和一半的位置 對函式做積分計算(圖形面積)，剛好可以保證，超過一半的部分，如果要拿掉一半的儲存令牌所需要的時間恰好是儲存同樣量（或者說是新令牌產生）時間花銷的兩倍，對應場景如果過了很長一段時間沒有使用(儲存的令牌會達到maxPermits),剛開始能接收請求的速率相對比較慢，然後再增長到穩定的消費速率
關鍵在於儲存的速率是和新令牌產生的速率一樣，但是消費的速率，當儲存的超過一半時，會慢於新令牌產生的速率，小於一半則速率是一樣的
TryAcquire
它會嘗試去獲取令牌，如果無法獲取就立即返回，否則再超時時間之內返回給定的令牌。 原始碼如下
public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
//1：使用微秒來轉換超時時間
long timeoutMicros = unit.toMicros(timeout);
checkPermits(permits);
long microsToWait;
synchronized (mutex) {
long nowMicros = readSafeMicros();
//2.1：如果下次能夠獲取令牌的時間超過超時時間範圍，立馬返回；
if (nextFreeTicketMicros > nowMicros + timeoutMicros) {
return false;
} else {
//2.2：獲取需要等待的時間，本次獲取的時間肯定不會超時
microsToWait = reserveNextTicket(permits, nowMicros);
}
}
//3：實行等待
ticker.sleepMicrosUninterruptibly(microsToWait);
return true;
}
複製程式碼
第一次執行的時候，nextFreeTicketMicros是建立時候的時間，必定小於當前時間，所以第一次肯定會放過，允許執行，只是需要計算要等待的時間。
private long reserveNextTicket(double requiredPermits, long nowMicros) {
//1：如果下次可以獲取令牌的時間在過去，更新
resync(nowMicros);
//2：計算距離下次獲取令牌需要的時間，如果nextFreeTikcetMicros>nowMicros，這個時間段必定在超時時間之內,假如入超時時間是0，那麼必定是microsToNextFreeTicket趨近於0，也就是立馬能夠放行；
long microsToNextFreeTicket = Math.max(0, nextFreeTicketMicros - nowMicros);
//3：計算需要消耗的儲存的令牌
double storedPermitsToSpend = Math.min(requiredPermits, this.storedPermits);
//4：計算需要新產生的令牌
double freshPermits = requiredPermits - storedPermitsToSpend;
//5：計算消耗儲存令牌所需要的時間和新產生令牌所需要的時間。對於Bursty來講，消耗儲存的令牌所需要時間為0，WarmingUp方式則是需要根據不同的場景有不同的結果
long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
+ (long) (freshPermits * stableIntervalMicros);
//6：下次能夠獲取令牌的時間，需要延遲當前已經等待的時間，也就是說，如果立馬有請求過來會放行，但是這個等待時間將會影響後續的請求訪問，也就是說，這次的請求如果當前的特別的多，下一次能夠請求的能夠允許的時間必定會有很長的延遲
this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;
//7：扣除消耗的儲存令牌
this.storedPermits -= storedPermitsToSpend;
//8：返回本次要獲取令牌所需要的時間,它肯定不會超過超時時間
return microsToNextFreeTicket;
}
複製程式碼
Acquire
它會阻塞知道允許放行，返回值為阻塞的時長 原始碼如下
public double acquire(int permits) {
long microsToWait = reserve(permits); //也就是呼叫reserveNextTicket
ticker.sleepMicrosUninterruptibly(microsToWait); //阻塞住需要等待的時長
return 1.0 * microsToWait / TimeUnit.SECONDS.toMicros(1L);
}
複製程式碼
TryAcquire 執行案例
程式設定10個執行緒,使得併發數為10，模擬線上的場景,任務內容如下
class MyTask implements Runnable{
private CountDownLatch latch;
private RateLimiter limiter;
public MyTask(CountDownLatch latch, RateLimiter limiter) {
this.latch = latch;
this.limiter = limiter;
}
@Overridepublic void run() {
try {
//使得執行緒同時觸發
latch.await();
System.out.println("time "+System.currentTimeMillis()+"ms :"+limiter.tryAcquire());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
複製程式碼
Bursty-TryAcquire
這裡設定限制每秒的流量為5，也就是說第一次請求過後，下次請求需要等200ms
RateLimiter r =RateLimiter.create(5);
ExecutorService service = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(10);
for (int i=0;i<10;i++)
{
service.submit(new MyTask(latch, r));
latch.countDown();
System.out.println("countdown:" + latch.getCount());
}
System.out.println("countdown over");
service.shutdown();
複製程式碼
結果如下
countdown:9 countdown:8 countdown:7 countdown:6 countdown:5 countdown:4 countdown:3 countdown:2 countdown:1 countdown:0 countdown over time 1538487195698ms :true time 1538487195699ms :false time 1538487195699ms :false time 1538487195699ms :false time 1538487195699ms :false time 1538487195699ms :false time 1538487195699ms :false time 1538487195698ms :false time 1538487195698ms :false time 1538487195699ms :false 複製程式碼
如果使得執行緒等待401ms,那麼程式會儲存的令牌為2個
注意剛開始儲存的時候，不是慢的，這裡的儲存量是慢慢增長，並且能夠立馬拿到
RateLimiter r =RateLimiter.create(5);
ExecutorService service = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(10);
for (int i=0;i<10;i++)
{
service.submit(new MyTask(latch, r));
if (i==9){
TimeUnit.MILLISECONDS.sleep(401);
System.out.println("sleep 10 seconds over");
}
latch.countDown();
System.out.println("countdown:" + latch.getCount());
}
System.out.println("countdown over");
service.shutdown();
複製程式碼
執行結果剛好允許3個執行
countdown:9 countdown:8 countdown:7 countdown:6 countdown:5 countdown:4 countdown:3 countdown:2 countdown:1 sleep 10 seconds over countdown:0 countdown over time 1538487297981ms :true time 1538487297981ms :false time 1538487297981ms :false time 1538487297981ms :false time 1538487297981ms :true time 1538487297981ms :true time 1538487297981ms :false time 1538487297981ms :false time 1538487297981ms :false time 1538487297981ms :false 複製程式碼
如果等待時間超過1秒，允許放行的流量也不會超過6個，儲存的令牌+第一個令牌
RateLimiter r =RateLimiter.create(5);
ExecutorService service = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(10);
for (int i=0;i<10;i++)
{
service.submit(new MyTask(latch, r));
if (i==9){
TimeUnit.MILLISECONDS.sleep(1001);
System.out.println("sleep 10 seconds over");
}
latch.countDown();
System.out.println("countdown:" + latch.getCount());
}
System.out.println("countdown over");
service.shutdown();
複製程式碼
結果為
countdown:9 countdown:8 countdown:7 countdown:6 countdown:5 countdown:4 countdown:3 countdown:2 countdown:1 sleep 10 seconds over countdown:0 countdown over time 1538487514780ms :true time 1538487514780ms :true time 1538487514780ms :true time 1538487514780ms :false time 1538487514780ms :true time 1538487514780ms :false time 1538487514780ms :false time 1538487514780ms :false time 1538487514780ms :true time 1538487514780ms :true 複製程式碼
WarmingUp-TryAcquire
使用warmingUp的方式由於預設已經儲存滿了令牌，那麼，它在第一次請求執行完之後，必須等待一定的時間才會讓下一次請求開始，而這個請求放行的時間則是會超過儲存所需要的時間
注意這裡的不同，預設是儲存滿的，也就是剛開始的消費要慢很多
RateLimiter r =RateLimiter.create(5,1,TimeUnit.SECONDS);
ExecutorService service = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(10);
for (int i=0;i<10;i++)
{
service.submit(new MyTask(latch, r));
latch.countDown();
System.out.println("countdown:" + latch.getCount());
}
System.out.println("countdown over");
service.shutdown();
複製程式碼
執行結果如下
countdown:9 countdown:8 countdown:7 countdown:6 countdown:5 countdown:4 countdown:3 countdown:2 countdown:1 countdown:0 countdown over time 1538487677462ms :true time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false time 1538487677462ms :false 複製程式碼
Acquire執行案例
所需要的task原始碼如下
class MyTask implements Runnable{
private CountDownLatch latch;
private RateLimiter limiter;
private long start;
public MyTask(CountDownLatch latch, RateLimiter limiter,long start) {
this.latch = latch;
this.limiter = limiter;
this.start=start;
}
@Overridepublic void run() {
try {
//使得執行緒同時觸發
latch.await();
System.out.printf("result:"+limiter.acquire(2));
System.out.println(" time "+(System.currentTimeMillis()-start)+"ms");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
複製程式碼
Busty-Acquire
Acquire會阻塞執行的結果，而且會提前消費
RateLimiter r =RateLimiter.create(1);
r.acquire();
System.out.println("time cost:"+(System.currentTimeMillis()-start)+"ms");
r.acquire();
System.out.println("time cost:"+(System.currentTimeMillis()-start)+"ms");
r.acquire(3);
System.out.println("time cost:"+(System.currentTimeMillis()-start)+"ms");
r.acquire();
System.out.println("time cost:"+(System.currentTimeMillis()-start)+"ms");
複製程式碼
第一次會立馬執行，然後因為請求了一次，下次發放令牌的時間往後遷移,獲取的令牌越多，下次能夠執行需要等待的時間越長
執行結果為
time cost:0ms time cost:1005ms time cost:2004ms time cost:5001ms 複製程式碼
在多執行緒背景執行如下
RateLimiter r =RateLimiter.create(1);
long start=System.currentTimeMillis();
r.acquire(3);
System.out.println("time cost:"+(System.currentTimeMillis()-start)+"ms");
ExecutorService service = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(10);
for (int i=0;i<10;i++)
{
service.submit(new MyTask(latch, r,start));
latch.countDown();
System.out.println("countdown:" + latch.getCount());
}
System.out.println("countdown over");
service.shutdown();
複製程式碼
結果如下
time cost:1ms countdown:9 countdown:8 countdown:7 countdown:6 countdown:5 countdown:4 countdown:3 countdown:2 countdown:1 countdown:0 countdown over result:2.995732 time 3024ms result:4.995725 time 5006ms result:6.995719 time 7007ms result:8.995716 time 9006ms result:10.995698 time 11004ms result:12.995572 time 13006ms result:14.995555 time 15007ms result:16.995543 time 17005ms result:18.995516 time 19005ms result:20.995463 time 21005ms 複製程式碼
WarmingUp-acquire
warmingUp通過acquire的方式獲取的令牌，同樣會被按照同步的方式獲取
RateLimiter r =RateLimiter.create(1,1,TimeUnit.SECONDS);
long start=System.currentTimeMillis();
r.acquire(3);
System.out.println("time cost:"+(System.currentTimeMillis()-start)+"ms”);
ExecutorService service = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(10);
for (int i=0;i<10;i++)
{
service.submit(new MyTask(latch, r,start));
latch.countDown();
System.out.println("countdown:" + latch.getCount());
}
System.out.println("countdown over");
service.shutdown();
複製程式碼
結果如下
time cost:0ms countdown:9 countdown:8 countdown:7 countdown:6 countdown:5 countdown:4 countdown:3 countdown:2 countdown:1 countdown:0 countdown over result:3.496859 time 3521ms result:5.496854 time 5506ms result:7.49685 time 7505ms result:9.496835 time 9504ms result:11.496821 time 11505ms result:13.496807 time 13502ms result:15.496793 time 15504ms result:17.496778 time 17506ms result:19.496707 time 19506ms result:21.496699 time 21506ms 複製程式碼
RateLimiter本身實現的就是一個令牌桶演算法