做業務的同學都知道，在現實情況中，往往會出現流量暴增的情況。這些流量可能來自於黑客的爬蟲，也可能來自於節日大促，或者其他一些渠道。當然業界都有對策，比如反爬、熔斷、降級、限流等等不一而足。
我們今天就來談談其中的限流。
先看看業界常用的限流組件：
-
單機版
- JDK自帶的鎖、信號量等
- Guava的RateLimiter
-
分布式
- Redis
- Redis-cell
- Redisson（基於lua腳本實現）
- Sentinel
- Hystrix
- Redis
這里我用單機和分布式的維度簡單划分了一下。其中有一些你應該見過。
那么什么時候用單機限流，什么時候用分布式限流呢？
其實要回答這個問題，首先要明確你的業務要限流的對象是什么。比如你的服務是單體的，那其實用單機限流正合適。當然現在這個時代的單體業務不多了，只有一些小項目用的比較多。還有一種情況就是不需要在業務層面精確限流，比如說我們的業務部署在多個機器或者容器上，對外以Http的形式暴露服務，並且在Nginx層做了一定的負載均衡，導致流量會比較均勻地分布到各台機器上，此時用單機限流也是不錯的：只要每台機器的流量都限制住，整體的流量就是被限制的。注意這里是近似，試想，如果我們的機器數量擴容了一倍，那整體的限流閾值就會增長一倍。所以說，單機限流閾值也常用來保護機器不被打崩，考慮的角度更多是機器的性能，而非細化到某個業務接口。
如果要精確限制某個業務接口的流量（服務暴露形式不限，可能是Http，也可能是RPC），在分布式部署的環境下就需要采用分布式限流的方案了。國內少數公司采用Netflix開源的Hystrix實現限流功能，還有一些公司直接引入了阿里開源的Sentinel來用，還有一些公司會使用Redisson提供的限流能力，或者直接編寫Lua腳本實現。當然，還有一些大廠會自研限流組件，更好地滿足自身業務需求。
那么問題來了，如果讓你設計一個限流組件，要考慮哪些因素呢？這里給出一些參考：
1、限流的維度是什么？這個問題直接決定了限流功能要在哪個層面來實現。比如針對Http接口，限流器可以配置在Nginx層，進行域名限流。作為面向用戶端防火牆的一個基礎組件。也可以下放到業務接口層，限制某個業務的流量。再進一步，如果我們要根據特定的接口參數進行限流（比如限制每個用戶在一段時間內請求某個接口的頻率），那限流組件就需要在接口層面實現了。
2、限流用什么算法？常見的有固定窗口法、漏斗法，以及令牌桶法，還有一些組件實現了具有預熱功能的算法。現實場景中，出於應對突發流量的考慮，令牌桶算法的應用更為廣泛。這個問題網上談的比較多，不再贅述了。我們這里貼一段Redisson中利用Lua腳本實現令牌桶算法的例子，感受一下：
## Redisson源碼片段[Java]：獲取令牌。
private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
"local rate = redis.call('hget', KEYS[1], 'rate');"
+ "local interval = redis.call('hget', KEYS[1], 'interval');"
+ "local type = redis.call('hget', KEYS[1], 'type');"
+ "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')"
+ "local valueName = KEYS[2];"
+ "local permitsName = KEYS[4];"
+ "if type == '1' then "
+ "valueName = KEYS[3];"
+ "permitsName = KEYS[5];"
+ "end;"
+ "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); "
+ "local currentValue = redis.call('get', valueName); "
+ "if currentValue ~= false then "
+ "local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "
+ "local released = 0; "
+ "for i, v in ipairs(expiredValues) do "
+ "local random, permits = struct.unpack('fI', v);"
+ "released = released + permits;"
+ "end; "
+ "if released > 0 then "
+ "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "
+ "currentValue = tonumber(currentValue) + released; "
+ "redis.call('set', valueName, currentValue);"
+ "end;"
+ "if tonumber(currentValue) < tonumber(ARGV[1]) then "
+ "local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), '+inf', 'withscores', 'limit', 0, 1); "
+ "return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval);"
+ "else "
+ "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); "
+ "redis.call('decrby', valueName, ARGV[1]); "
+ "return nil; "
+ "end; "
+ "else "
+ "redis.call('set', valueName, rate); "
+ "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); "
+ "redis.call('decrby', valueName, ARGV[1]); "
+ "return nil; "
+ "end;",
Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),
value, System.currentTimeMillis(), ThreadLocalRandom.current().nextLong());
3、限流數據要保存在哪里？這個問題的答案依賴於具體的實現方案。比如我們如果用基於Redis發展出來的組件（比如Redis-cell、Redisson）來實現，限流數據就是存在Redis服務器中的。而如果采用Sentinel實現，限流數據就是存在內存中的。
4、限流數據量的控制。這個問題的解決方案依賴於限流對象的數量。如果是針對有限的幾個接口做限流，數據量小到幾乎可以不用考慮。但如果是前面提到的“根據特定的業務參數進行限流”這種場景，就可能出現問題：比如針對用戶ID做限流，那可能需要保存對應量級的限流數據（每個正在訪問的用戶都要記錄訪問頻次）。如果設計不恰當的話，內存很快就會上漲甚至被打爆。不信的話我們可以估算一下，如果用Redis實現，結合了業務屬性的Redis keys一般要占用幾百字節左右，那么1千萬個用戶就需要占用幾個GB的空間。如果換到保存在內存中，一條請求用的限流對象同樣可能也要占用幾百字節。如果我們放任這些數據無限增加的話，后果可能是災難性的。所以你一定想到了應對辦法，那就是要對老數據做過期刪除處理。具體到實現的話，Redis可以設置keys的過期時間，讓老keys過期后自動刪除。內存中可以設置最多保存多少條限流數據，超過閾值時觸發老數據淘汰機制，最常用的是LRU算法。當然，過期時間或者內存容量上限，都需要根據業務實際情況進行制定。PS：實際上Sentinel就利用了Google開源的ConcurrentLinkedHashMap，利用它實現了LRU：
## Sentinel源碼片段[Java]：利用ConcurrentLinkedHashMap實現LRU，可以淘汰老數據。
public class ConcurrentLinkedHashMapWrapper<T, R> implements CacheMap<T, R> {
private final ConcurrentLinkedHashMap<T, R> map;
public ConcurrentLinkedHashMapWrapper(long size) {
if (size <= 0) {
throw new IllegalArgumentException("Cache max capacity should be positive: " + size);
}
this.map = new ConcurrentLinkedHashMap.Builder<T, R>()
.concurrencyLevel(DEFAULT_CONCURRENCY_LEVEL)
.maximumWeightedCapacity(size)
.weigher(Weighers.singleton())
.build();
}
...
}
5、時鍾回撥情況的處理。這個問題一般在精細限流的場景更容易出現，比如限制幾個毫秒內只能通過幾個請求。一旦NTP服務器同步出現抖動，或者服務器本地時間被人為修改，就可能會導致獲取令牌算法出現錯誤，進而導致限流算法失效。常見的解決思路是讓限流服務器重新獲取一次時間，避免多個請求端的時間不一致。Redis-cell源碼中就采用了這一方案：
## Redis-cell源碼[rust]：獲取令牌前，主動同步當前時間。
/// Gets the given key's value and the current time as dictated by the
/// store (this is done so that rate limiters running on a variety of
/// different nodes can operate with a consistent clock instead of using
/// their own). If the key was unset, -1 is returned.
fn get_with_time(&self, key: &str) -> Result<(i64, time::Tm), CellError>;
fn get_with_time(&self, key: &str) -> Result<(i64, time::Tm), CellError> {
match self.map.get(key) {
Some(n) => Ok((*n, time::now_utc())),
None => Ok((-1, time::now_utc())),
}
}
// 下面的代碼是rust庫函數，獲取時間
/// Returns the current time in UTC
pub fn now_utc() -> Tm {
at_utc(get_time())
}
/**
* Returns the current time as a `timespec` containing the seconds and
* nanoseconds since 1970-01-01T00:00:00Z.
*/
pub fn get_time() -> Timespec {
let (sec, nsec) = sys::get_time();
Timespec::new(sec, nsec)
}
pub fn get_time() -> (i64, i32) {
let mut tv = libc::timespec { tv_sec: 0, tv_nsec: 0 };
unsafe { libc::clock_gettime(libc::CLOCK_REALTIME, &mut tv); }
(tv.tv_sec as i64, tv.tv_nsec as i32)
}
從以上幾點來看，要想設計一個好用且合格的限流組件，還真不是件容易的事情。
最后，本文也是實際工作中的一些經驗之談，很多細節尚沒有講到。有興趣的同學可繼續深究各種限流組件的實現原理，如有其他觀點，歡迎交流。
嘿嘿，給自己的公眾號打個廣告，歡迎大家關注我的公眾號：xiaoxi666，一起來玩一起學！
