來，四種分布式限流算法實現(xiàn)！
大家好，我是老三，最近公司在搞年終大促，隨著各種營銷活動“組合拳”打出，進(jìn)站流量時不時會有一個小波峰，一般情況下，當(dāng)然是流量越多越好，前提是系統(tǒng)能杠地住。大家都知道，一個分布式系統(tǒng)，有兩個“棄車保帥”的策略：限流和熔斷，這期，我們就來討論一下分布式系統(tǒng)的限流。
探探限流
帶著問題走近限流
為什么要限流呢？
就像我上面說的，流量多，的確是一件好事，但是如果過載，把系統(tǒng)打掛了，那大家都要吃席了。
沒逝吧
所以，在各種大促活動之前，要對系統(tǒng)進(jìn)行壓測，評估整個系統(tǒng)的峰值QPS，要做一些限流的設(shè)置，超過一定閾值，就拒絕處理或者延后處理，避免把系統(tǒng)打掛的情況出現(xiàn)。
限流和熔斷有什么區(qū)別？
限流發(fā)生在流量進(jìn)來之前，超過的流量進(jìn)行限制。
熔斷是一種應(yīng)對故障的機(jī)制，發(fā)生在流量進(jìn)來之后，如果系統(tǒng)發(fā)生故障或者異常，熔斷會自動切斷請求，防止故障進(jìn)一步擴(kuò)展，導(dǎo)致服務(wù)雪崩。
限流和削峰有什么區(qū)別？
削峰是對流量的平滑處理，通過緩慢地增加請求的處理速率來避免系統(tǒng)瞬時過載。
削峰大概就是水庫，把流量儲存起來，慢慢流，限流大概就是閘口，拒絕超出的流量。
限流的通用流程
那么具體限流怎么實現(xiàn)呢？可以概括為以下幾個步驟：
限流通用流程
- 統(tǒng)計請求流量：記錄請求的數(shù)量或速率，可以通過計數(shù)器、滑動窗口等方式進(jìn)行統(tǒng)計。
- 判斷是否超過限制：根據(jù)設(shè)定的限制條件，判斷當(dāng)前請求流量是否超過限制。
- 執(zhí)行限流策略：如果請求流量超過限制，執(zhí)行限流策略，如拒絕請求、延遲處理、返回錯誤信息等。
- 更新統(tǒng)計信息：根據(jù)請求的處理結(jié)果，更新統(tǒng)計信息，如增加計數(shù)器的值、更新滑動窗口的數(shù)據(jù)等。
- 重復(fù)執(zhí)行以上步驟：不斷地統(tǒng)計請求流量、判斷是否超過限制、執(zhí)行限流策略、更新統(tǒng)計信息
需要注意的是，具體的限流算法實現(xiàn)可能會根據(jù)不同的場景和需求進(jìn)行調(diào)整和優(yōu)化，比如使用令牌桶算法、漏桶算法等。
單機(jī)限流和分布式限流
我們注意到，在限流的通用流程里，需要統(tǒng)計請求量、更新統(tǒng)計量，那么這個請求量的統(tǒng)計和更新就必須維護(hù)在一個存儲里。
假如只是一個單機(jī)版的環(huán)境，那就很好辦了，直接儲存到本地。
單機(jī)vs集群
但是一般來講，我們的服務(wù)都是集群部署的，如何來實現(xiàn)多臺機(jī)器之間整體的限流呢？
這時候就可以把我們的統(tǒng)計信息放到Tair或Redis等分布式的K-V存儲中。
四種限流算法與分布式實現(xiàn)
接下來，我們開始實現(xiàn)一些常見的限流算法，這里使用Redis作為分布式存儲，Redis不用多說了吧，最流行的分布式緩存DB；Redission作為Redis客戶端，Redission單純只是用來做分布式鎖，有些”屈才“，其實用來作為Redis的客戶端也非常好用。
五種限流算法分布式實現(xiàn)
在開始之前，我們先簡單準(zhǔn)備一下環(huán)境，Redis安裝和項目創(chuàng)建就不多說了。
- 添加依賴
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.2</version>
</dependency>- 用單例模式獲取RedissonClient，這里就不注冊成bean了，跑單測太慢
public class RedissonConfig {
private static final String REDIS_ADDRESS = "redis://127.0.0.1:6379";
private static volatile RedissonClient redissonClient;
public static RedissonClient getInstance(){
if (redissnotallow==null){
synchronized (RedissonConfig.class){
if (redissnotallow==null){
Config config = new Config();
config.useSingleServer().setAddress(REDIS_ADDRESS);
redissonClient = Redisson.create(config);
return redissonClient;
}
}
}
return redissonClient;
}
}固定窗口限流算法
算法原理
固定窗口算法，很多參考資料也稱之為計數(shù)器算法，當(dāng)然我個人理解，計數(shù)器算法是固定窗口算法的一種特例，當(dāng)然我們不糾結(jié)那么多。
固定窗口算法，是一種比較簡單的限流算法，它把時間劃分為固定的時間窗口，每個窗口內(nèi)允許的請求次數(shù)設(shè)置限制。如果在一個時間窗口內(nèi)，請求次數(shù)超過了上限，那么就會觸發(fā)限流。
在這里插入圖片描述
算法實現(xiàn)
基于Redisson的實現(xiàn)固定窗口相當(dāng)簡單。在每個窗口期內(nèi)，我們可以通過incrementAndGet操作來統(tǒng)計請求的數(shù)量。一旦窗口期結(jié)束，我們可以利用Redis的鍵過期功能來自動重置計數(shù)。
- 來看下代碼實現(xiàn)：
public class FixedWindowRateLimiter {
public static final String KEY = "fixedWindowRateLimiter:";
/**
* 請求限制數(shù)量
*/
private Long limit;
/**
* 窗口大小（單位：S）
*/
private Long windowSize;
public FixedWindowRateLimiter(Long limit, Long windowSize) {
this.limit = limit;
this.windowSize = windowSize;
}
/**
* 固定窗口限流
*/
public boolean triggerLimit(String path) {
RedissonClient redissonClient = RedissonConfig.getInstance();
//加分布式鎖，防止并發(fā)情況下窗口初始化時間不一致問題
RLock rLock = redissonClient.getLock(KEY + "LOCK:" + path);
try {
rLock.lock(100, TimeUnit.MILLISECONDS);
String redisKey = KEY + path;
RAtomicLong counter = redissonClient.getAtomicLong(redisKey);
//計數(shù)
long count = counter.incrementAndGet();
//如果為1的話，就說明窗口剛初始化
if (count == 1) {
//直接設(shè)置過期時間，作為窗口
counter.expire(windowSize, TimeUnit.SECONDS);
}
//觸發(fā)限流
if (count > limit) {
//觸發(fā)限流的不記在請求數(shù)量中
counter.decrementAndGet();
return true;
}
return false;
} finally {
rLock.unlock();
}
}
}這里還額外用了一個分布式鎖，來解決并發(fā)情況下，窗口的初始化問題。
- 再來測試一下
class FixedWindowRateLimiterTest {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(20, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
@Test
@DisplayName("1min限制10次請求固定窗口測試")
void triggerLimit() throws InterruptedException {
FixedWindowRateLimiter fixedWindowRateLimiter = new FixedWindowRateLimiter(10L,60L);
//模擬不同窗口內(nèi)的調(diào)用
for (int i = 0; i < 3; i++) {
CountDownLatch countDownLatch = new CountDownLatch(20);
//20個線程并發(fā)調(diào)用
for (int j = 0; j < 20; j++) {
threadPoolExecutor.execute(() -> {
boolean isLimit = fixedWindowRateLimiter.triggerLimit("/test");
System.out.println(isLimit);
countDownLatch.countDown();
});
}
countDownLatch.await();
//休眠1min
TimeUnit.MINUTES.sleep(1);
}
}
}當(dāng)然大家也可以寫個接口，用Jmeter之類的壓測工具來進(jìn)行測試。
固定窗口算法的優(yōu)點是實現(xiàn)簡單，占用空間小，但是它存在臨界問題，由于窗口的切換是瞬間完成的，因此請求的處理并不平滑，可能會在窗口切換的瞬間出現(xiàn)流量的劇烈波動。
比如這個例子，假如在00:02，突然有大量請求過來，但是我們這時候計數(shù)重置了，那么就沒法限制突發(fā)的這些流量。
臨界值問題
滑動窗口算法
為了緩解固定窗口的突發(fā)流量問題，可以采用滑動窗口算法，計算機(jī)網(wǎng)絡(luò)中TCP的流量控制就是采用滑動窗口算法。
算法原理
滑動窗口限流算法的原理是將一個大的時間窗口劃分為多個小的時間窗口，每個小的窗口都有獨立的計數(shù)。
請求過來的時候，判斷請求的次數(shù)是否超過整個窗口的限制。窗口的移動是每次向前滑動一個小的單元窗口。
例如下面這個滑動窗口，將大時間窗口1min分成了5個小窗口，每個小窗口的時間是12s。
每個單元格有自己獨立的計數(shù)器，每過12s就會向前移動一格。
假如有請求在00:01的時候過來，這時候窗口的計數(shù)就是3+12+9+15=39，也能起到限流的作用。
滑動窗口算法示意圖
這就是為什么滑動窗口能解決臨界問題，滑的格子越多，那么整體的滑動就會越平滑,限流的效果就會越精準(zhǔn)。
算法實現(xiàn)
那么我們這里怎么實現(xiàn)滑動窗口限流算法呢？非常簡單，我們可以直接使用Redis的有序集合（zset）結(jié)構(gòu)。
我們使用時間戳作為score和member，有請求過來的時候，就把當(dāng)前時間戳添加到有序集合里。那么窗口之外的請求，我們可以根據(jù)窗口大小，計算出起始時間戳，刪除窗口外的請求。這樣，有序集合的大小，就是我們這個窗口的請求數(shù)了。
zset實現(xiàn)滑動窗口
- 代碼實現(xiàn)
public class SlidingWindowRateLimiter {
public static final String KEY = "slidingWindowRateLimiter:";
/**
* 請求次數(shù)限制
*/
private Long limit;
/**
* 窗口大小（單位：S）
*/
private Long windowSize;
public SlidingWindowRateLimiter(Long limit, Long windowSize) {
this.limit = limit;
this.windowSize = windowSize;
}
public boolean triggerLimit(String path) {
RedissonClient redissonClient = RedissonConfig.getInstance();
//窗口計數(shù)
RScoredSortedSet<Long> counter = redissonClient.getScoredSortedSet(KEY + path);
//使用分布式鎖，避免并發(fā)設(shè)置初始值的時候，導(dǎo)致窗口計數(shù)被覆蓋
RLock rLock = redissonClient.getLock(KEY + "LOCK:" + path);
try {
rLock.lock(200, TimeUnit.MILLISECONDS);
// 當(dāng)前時間戳
long currentTimestamp = System.currentTimeMillis();
// 窗口起始時間戳
long windowStartTimestamp = currentTimestamp - windowSize * 1000;
// 移除窗口外的時間戳，左閉右開
counter.removeRangeByScore(0, true, windowStartTimestamp, false);
// 將當(dāng)前時間戳作為score,也作為member，
// TODO:高并發(fā)情況下可能沒法保證唯一，可以加一個唯一標(biāo)識
counter.add(currentTimestamp, currentTimestamp);
//使用zset的元素個數(shù)，作為請求計數(shù)
long count = counter.size();
// 判斷時間戳數(shù)量是否超過限流閾值
if (count > limit) {
System.out.println("[triggerLimit] path:" + path + " count:" + count + " over limit:" + limit);
return true;
}
return false;
} finally {
rLock.unlock();
}
}
}這里還有一個小的可以完善的點，zset在member相同的情況下，是會覆蓋的，也就是說高并發(fā)情況下，時間戳可能會重復(fù)，那么就有可能統(tǒng)計的請求偏少，這里可以用時間戳+隨機(jī)數(shù)來緩解，也可以生成唯一序列來解決，比如UUID、雪花算法等等。
- 還是來測試一下
class SlidingWindowRateLimiterTest {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
@Test
@DisplayName("滑動窗口限流")
void triggerLimit() throws InterruptedException {
SlidingWindowRateLimiter slidingWindowRateLimiter = new SlidingWindowRateLimiter(10L, 1L);
//模擬在不同時間片內(nèi)的請求
for (int i = 0; i < 8; i++) {
CountDownLatch countDownLatch = new CountDownLatch(20);
for (int j = 0; j < 20; j++) {
threadPoolExecutor.execute(() -> {
boolean isLimit = slidingWindowRateLimiter.triggerLimit("/test");
System.out.println(isLimit);
countDownLatch.countDown();
});
}
countDownLatch.await();
//休眠10s
TimeUnit.SECONDS.sleep(10L);
}
}
}用Redis實現(xiàn)了滑動窗口限流，解決了固定窗口限流的邊界問題，當(dāng)然這里也帶來了新的問題，因為我們存儲了窗口期的所有請求，所以高并發(fā)的情況下，可能會比較占內(nèi)存。
漏桶算法
我們可以看到，計數(shù)器類的限流，體現(xiàn)的是一個“戛然而止”，超過限制，立馬決絕，但是有時候，我們可能只是希望請求平滑一些，追求的是“波瀾不驚”，這時候就可以考慮使用其它的限流算法。
算法原理
漏桶算法（Leaky Bucket），名副其實，就是請求就像水一樣以任意速度注入漏桶，而桶會按照固定的速率將水漏掉。
漏桶算法
當(dāng)進(jìn)水速率大于出水速率的時候，漏桶會變滿，此時新進(jìn)入的請求將會被丟棄。
漏桶算法的兩大作用是網(wǎng)絡(luò)流量整形（Traffic Shaping）和速度限制（Rate Limiting）。
算法實現(xiàn)
我們接著看看具體應(yīng)該怎么實現(xiàn)。
在滑動窗口限流算法里我們用到了RScoredSortedSet，非常好用對不對，這里也可以用這個結(jié)構(gòu)，直接使用ZREMRANGEBYSCORE命令來刪除舊的請求。
進(jìn)水就不用多說了，請求進(jìn)來，判斷桶有沒有滿，滿了就拒絕，沒滿就往桶里丟請求。
那么出水怎么辦呢？得保證穩(wěn)定速率出水，可以用一個定時任務(wù)，來定時去刪除舊的請求。
- 代碼實現(xiàn)
public class LeakyBucketRateLimiter {
private RedissonClient redissonClient = RedissonConfig.getInstance();
private static final String KEY_PREFIX = "LeakyBucket:";
/**
* 桶的大小
*/
private Long bucketSize;
/**
* 漏水速率，單位:個/秒
*/
private Long leakRate;
public LeakyBucketRateLimiter(Long bucketSize, Long leakRate) {
this.bucketSize = bucketSize;
this.leakRate = leakRate;
//這里啟動一個定時任務(wù)，每s執(zhí)行一次
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
executorService.scheduleAtFixedRate(this::leakWater, 0, 1, TimeUnit.SECONDS);
}
/**
* 漏水
*/
public void leakWater() {
RSet<String> pathSet=redissonClient.getSet(KEY_PREFIX+":pathSet");
//遍歷所有path,刪除舊請求
for(String path:pathSet){
String redisKey = KEY_PREFIX + path;
RScoredSortedSet<Long> bucket = redissonClient.getScoredSortedSet(KEY_PREFIX + path);
// 獲取當(dāng)前時間
long now = System.currentTimeMillis();
// 刪除舊的請求
bucket.removeRangeByScore(0, true,now - 1000 * leakRate,true);
}
}
/**
* 限流
*/
public boolean triggerLimit(String path) {
//加鎖，防止并發(fā)初始化問題
RLock rLock = redissonClient.getLock(KEY_PREFIX + "LOCK:" + path);
try {
rLock.lock(100,TimeUnit.MILLISECONDS);
String redisKey = KEY_PREFIX + path;
RScoredSortedSet<Long> bucket = redissonClient.getScoredSortedSet(redisKey);
//這里用一個set，來存儲所有path
RSet<String> pathSet=redissonClient.getSet(KEY_PREFIX+":pathSet");
pathSet.add(path);
// 獲取當(dāng)前時間
long now = System.currentTimeMillis();
// 檢查桶是否已滿
if (bucket.size() < bucketSize) {
// 桶未滿，添加一個元素到桶中
bucket.add(now,now);
return false;
}
// 桶已滿，觸發(fā)限流
System.out.println("[triggerLimit] path:"+path+" bucket size:"+bucket.size());
return true;
}finally {
rLock.unlock();
}
}
}在代碼實現(xiàn)里，我們用了RSet來存儲path，這樣一來，一個定時任務(wù)，就可以搞定所有path對應(yīng)的桶的出水，而不用每個桶都創(chuàng)建一個一個定時任務(wù)。
這里我直接用ScheduledExecutorService啟動了一個定時任務(wù)，1s跑一次，當(dāng)然集群環(huán)境下，每臺機(jī)器都跑一個定時任務(wù)，對性能是極大的浪費，而且不好管理，我們可以用分布式定時任務(wù)，比如xxl-job去執(zhí)行l(wèi)eakWater。
- 最后還是大家熟悉的測試
class LeakyBucketRateLimiterTest {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
@Test
@DisplayName("漏桶算法")
void triggerLimit() throws InterruptedException {
LeakyBucketRateLimiter leakyBucketRateLimiter = new LeakyBucketRateLimiter(10L, 1L);
for (int i = 0; i < 8; i++) {
CountDownLatch countDownLatch = new CountDownLatch(20);
for (int j = 0; j < 20; j++) {
threadPoolExecutor.execute(() -> {
boolean isLimit = leakyBucketRateLimiter.triggerLimit("/test");
System.out.println(isLimit);
countDownLatch.countDown();
});
}
countDownLatch.await();
//休眠10s
TimeUnit.SECONDS.sleep(10L);
}
}
}漏桶算法能夠有效防止網(wǎng)絡(luò)擁塞，實現(xiàn)也比較簡單。
但是，因為漏桶的出水速率是固定的，假如突然來了大量的請求，那么只能丟棄超量的請求，即使下游能處理更大的流量，沒法充分利用系統(tǒng)資源。
令牌桶算法
令牌桶算法來了！
算法原理
令牌桶算法是對漏桶算法的一種改進(jìn)。
它的主要思想是：系統(tǒng)以一種固定的速率向桶中添加令牌，每個請求在發(fā)送前都需要從桶中取出一個令牌，只有取到令牌的請求才被通過。因此，令牌桶算法允許請求以任意速率發(fā)送，只要桶中有足夠的令牌。
令牌桶算法
算法實現(xiàn)
我們繼續(xù)看怎么實現(xiàn)，首先是要發(fā)放令牌，要固定速率，那我們又得開個線程，定時往桶里投令牌，然后……
——然后Redission提供了令牌桶算法的實現(xiàn)，舒不舒服？
拿來吧你
拿來就用！
- 代碼實現(xiàn)
public class TokenBucketRateLimiter {
public static final String KEY = "TokenBucketRateLimiter:";
/**
* 閾值
*/
private Long limit;
/**
* 添加令牌的速率，單位：個/秒
*/
private Long tokenRate;
public TokenBucketRateLimiter(Long limit, Long tokenRate) {
this.limit = limit;
this.tokenRate = tokenRate;
}
/**
* 限流算法
*/
public boolean triggerLimit(String path){
RedissonClient redissnotallow=RedissonConfig.getInstance();
RRateLimiter rateLimiter = redissonClient.getRateLimiter(KEY+path);
// 初始化，設(shè)置速率模式，速率，間隔，間隔單位
rateLimiter.trySetRate(RateType.OVERALL, limit, tokenRate, RateIntervalUnit.SECONDS);
// 獲取令牌
return rateLimiter.tryAcquire();
}
}Redisson實現(xiàn)的，還是比較穩(wěn)的，這里就不測試了。
關(guān)于Redission是怎么實現(xiàn)這個限速器的，大家可以看一下參考[3]，還是Redisson家的老傳統(tǒng)——Lua腳本，設(shè)計相當(dāng)巧妙。
總結(jié)
在這篇文章里，我們對四（三）種限流算法進(jìn)行了分布式實現(xiàn)，采用了非常好用的Redission客戶端，當(dāng)然我們也有不完善的地方：
- 并發(fā)處理采用了分布式鎖，高并發(fā)情況下，對性能有一定損耗，邏輯最好還是直接采用Lua腳本實現(xiàn)，來提高性能
- 可以提供更加優(yōu)雅的調(diào)用方式，比如利用aop實現(xiàn)注解式調(diào)用，代碼設(shè)計也可以更加優(yōu)雅，繼承體系可以完善一下
- 沒有實現(xiàn)限流的拒絕策略，比如拋異常、緩存、丟進(jìn)MQ打散……限流是一種方法，最終的目的還是盡可能保證系統(tǒng)平穩(wěn)
如果后面有機(jī)會，希望可以繼續(xù)完善這個簡單的Demo，達(dá)到工程級的應(yīng)用。
除此之外，市面上也有很多好用的開源限流工具：
- Guava RateLimiter ，基于令牌桶算法限流，當(dāng)然是單機(jī)的；
- Sentinel ，基于滑動窗口限流，支持單機(jī)，也支持集群
- 網(wǎng)關(guān)限流，很多網(wǎng)關(guān)自帶限流方法，比如Spring Cloud Gateway、Nginx
……
好了，這期文章就到這里了，我們下期見。
參考：
[1]. 面試官：來，年輕人！請手?jǐn)]5種常見限流算法！
[2].https://zhuanlan.zhihu.com/p/479956069
[3].https://github.com/oneone1995/blog/issues/13
