來談談限流-RateLimiter原始碼分析
前一篇文章
提到了限流的幾種常見演算法，本文將分析guava限流類RateLimiter
的實現。
RateLimiter
有兩個實現類：SmoothBursty
和SmoothWarmingUp
，其都是令牌桶演算法的變種實現，區別在於SmoothBursty
加令牌的速度是恆定的，而SmoothWarmingUp
會有個預熱期，在預熱期內加令牌的速度是慢慢增加的，直到達到固定速度為止。其適用場景是，對於有的系統而言剛啟動時能承受的QPS較小，需要預熱一段時間後才能達到最佳狀態。
更多文章見個人部落格：github.com/farmerjohng…
基本使用
RateLimiter
的使用很簡單：
//create方法傳入的是每秒生成令牌的個數
RateLimiter rateLimiter= RateLimiter.create(1);
for (int i = 0; i < 5; i++) {
//acquire方法傳入的是需要的令牌個數，當令牌不足時會進行等待，該方法返回的是等待的時間
double waitTime=rateLimiter.acquire(1);
System.out.println(System.currentTimeMillis()/1000+" , "+waitTime);
}
複製程式碼
輸出如下：
1548070953 , 0.0 1548070954 , 0.998356 1548070955 , 0.998136 1548070956 , 0.99982 複製程式碼
需要注意的是，當令牌不足時，acquire
方法並不會阻塞本次呼叫，而是會算在下次呼叫的頭上
。比如第一次呼叫時，令牌桶中並沒有令牌，但是第一次呼叫也沒有阻塞，而是在第二次呼叫的時候阻塞了1秒。也就是說，每次呼叫欠的令牌（如果桶中令牌不足）都是讓下一次呼叫買單
。
RateLimiter rateLimiter= RateLimiter.create(1); double waitTime=rateLimiter.acquire(1000); System.out.println(System.currentTimeMillis()/1000+" , "+waitTime); waitTime=rateLimiter.acquire(1); System.out.println(System.currentTimeMillis()/1000+" , "+waitTime); 複製程式碼
輸出如下：
1548072250 , 0.0 1548073250 , 999.998773 複製程式碼
這樣設計的目的是：
Last, but not least: consider a RateLimiter with rate of 1 permit per second, currently completely unused, and an expensive acquire(100) request comes. It would be nonsensical to just wait for 100 seconds, and /then/ start the actual task. Why wait without doing anything? A much better approach is to /allow/ the request right away (as if it was an acquire(1) request instead), and postpone /subsequent/ requests as needed. In this version, we allow starting the task immediately, and postpone by 100 seconds future requests, thus we allow for work to get done in the meantime instead of waiting idly. 複製程式碼
簡單的說就是，如果每次請求都為本次買單會有不必要的等待。比如說令牌增加的速度為每秒1個，初始時桶中沒有令牌，這時來了個請求需要100個令牌，那需要等待100s後才能開始這個任務。所以更好的辦法是先放行這個請求，然後延遲之後的請求。
另外，RateLimiter還有個tryAcquire
方法，如果令牌夠會立即返回true，否則立即返回false。
原始碼分析
本文主要分析SmoothBursty
的實現。
首先看SmoothBursty
中的幾個關鍵欄位:
// 桶中最多存放多少秒的令牌數 final double maxBurstSeconds; //桶中的令牌個數 double storedPermits; //桶中最多能存放多少個令牌，=maxBurstSeconds*每秒生成令牌個數 double maxPermits; //加入令牌的平均間隔，單位為微秒，如果加入令牌速度為每秒5個，則該值為1000*1000/5 double stableIntervalMicros; //下一個請求需要等待的時間 private long nextFreeTicketMicros = 0L; 複製程式碼
RateLimiter的建立
先看建立RateLimiter的create方法。
// permitsPerSecond為每秒生成的令牌數
public static RateLimiter create(double permitsPerSecond) {
return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
}
//SleepingStopwatch主要用於計時和休眠
static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
//建立一個SmoothBursty
RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
rateLimiter.setRate(permitsPerSecond);
return rateLimiter;
}
複製程式碼
create方法主要就是建立了一個SmoothBursty
例項，並呼叫了其setRate
方法。注意這裡的maxBurstSeconds
寫死為1.0。
@Override
final void doSetRate(double permitsPerSecond, long nowMicros) {
resync(nowMicros);
double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
this.stableIntervalMicros = stableIntervalMicros;
doSetRate(permitsPerSecond, stableIntervalMicros);
}
void resync(long nowMicros) {
// 如果當前時間比nextFreeTicketMicros大，說明上一個請求欠的令牌已經補充好了，本次請求不用等待
if (nowMicros > nextFreeTicketMicros) {
// 計算這段時間內需要補充的令牌，coolDownIntervalMicros返回的是stableIntervalMicros
double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
// 更新桶中的令牌，不能超過maxPermits
storedPermits = min(maxPermits, storedPermits + newPermits);
// 這裡先設定為nowMicros
nextFreeTicketMicros = nowMicros;
}
}
@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 {
//第一次呼叫oldMaxPermits為0，所以storedPermits（桶中令牌個數）也為0
storedPermits =
(oldMaxPermits == 0.0)
? 0.0 // initial state
: storedPermits * maxPermits / oldMaxPermits;
}
}
複製程式碼
setRate
方法中設定了maxPermits=maxBurstSeconds * permitsPerSecond
;而maxBurstSeconds
為1，所以maxBurstSeconds
只會儲存1秒中的令牌數。
需要注意的是SmoothBursty
是非public的類，也就是說只能通過RateLimiter.create
方法建立，而該方法中的maxBurstSeconds
是寫死1.0的，也就是說我們只能建立桶大小為permitsPerSecond*1的SmoothBursty
物件（當然反射的方式不在討論範圍），在guava的github倉庫裡有好幾條issue（issue1
,issue2
,issue3
,issue4
）希望能由外部設定maxBurstSeconds
，但是並沒有看到官方人員的回覆。而在唯品會的開源專案vjtools
中，有人提出了這個問題
，唯品會的同學對guava的RateLimiter進行了拓展
。
對於guava的這樣設計我很不理解，有清楚的朋友可以說下~
到此為止一個SmoothBursty
物件就建立好了，接下來我們分析其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);
}
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
// 這裡呼叫了上面提到的resync方法，可能會更新桶中的令牌值和nextFreeTicketMicros
resync(nowMicros);
// 如果上次請求花費的令牌還沒有補齊，這裡returnValue為上一次請求後需要等待的時間，否則為nowMicros
long returnValue = nextFreeTicketMicros;
double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
// 缺少的令牌數
double freshPermits = requiredPermits - storedPermitsToSpend;
// waitMicros為下一次請求需要等待的時間；SmoothBursty的storedPermitsToWaitTime返回0
long waitMicros =
storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
+ (long) (freshPermits * stableIntervalMicros);
// 更新nextFreeTicketMicros
this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
// 減少令牌
this.storedPermits -= storedPermitsToSpend;
return returnValue;
}
複製程式碼
acquire
中會呼叫reserve
方法獲得當前請求需要等待的時間，然後進行休眠。reserve
方法最終會呼叫到reserveEarliestAvailable
，在該方法中會先呼叫上文提到的resync
方法對桶中的令牌進行補充（如果需要的話），然後減少桶中的令牌，以及計算這次請求欠的令牌數及需要等待的時間（由下次請求負責等待）。
如果上一次請求沒有欠令牌或欠的令牌已經還清則返回值為nowMicros
，否則返回值為上一次請求缺少的令牌個數*生成一個令牌所需要的時間。