蝦皮二面：常見的限流算法有哪些？

你好，我是 Guide。分享一道群友面試蝦皮遇到的關(guān)于限流的面試真題。

下面是最近更新的一些面試真題：

1. 限流

限流顧名思義，就是對請求或并發(fā)數(shù)進(jìn)行限制；通過對一個時間窗口內(nèi)的請求量進(jìn)行限制來保障系統(tǒng)的正常運(yùn)行。如果我們的服務(wù)資源有限、處理能力有限，就需要對調(diào)用我們服務(wù)的上游請求進(jìn)行限制，以防止自身服務(wù)由于資源耗盡而停止服務(wù)。

在限流中有兩個概念需要了解。

• 閾值：在一個單位時間內(nèi)允許的請求量。如 QPS 限制為10，說明 1 秒內(nèi)最多接受 10 次請求。
• 拒絕策略：超過閾值的請求的拒絕策略，常見的拒絕策略有直接拒絕、排隊等待等。

2. 固定窗口算法

固定窗口算法又叫計數(shù)器算法，是一種簡單方便的限流算法。主要通過一個支持原子操作的計數(shù)器來累計 1 秒內(nèi)的請求次數(shù)，當(dāng) 1 秒內(nèi)計數(shù)達(dá)到限流閾值時觸發(fā)拒絕策略。每過 1 秒，計數(shù)器重置為 0 開始重新計數(shù)。

2.1. 代碼實現(xiàn)

下面是簡單的代碼實現(xiàn)，QPS 限制為 2，這里的代碼做了一些優(yōu)化，并沒有單獨開一個線程去每隔 1 秒重置計數(shù)器，而是在每次調(diào)用時進(jìn)行時間間隔計算來確定是否先重置計數(shù)器。

/**
?*?@author?https://www.wdbyte.com
?*/
public?class?RateLimiterSimpleWindow?{
????//?閾值
????private?static?Integer?QPS?=?2;
????//?時間窗口（毫秒）
????private?static?long?TIME_WINDOWS?=?1000;
????//?計數(shù)器
????private?static?AtomicInteger?REQ_COUNT?=?new?AtomicInteger();
????
????private?static?long?START_TIME?=?System.currentTimeMillis();

????public?synchronized?static?boolean?tryAcquire()?{
????????if?((System.currentTimeMillis()?-?START_TIME)?>?TIME_WINDOWS)?{
????????????REQ_COUNT.set(0);
????????????START_TIME?=?System.currentTimeMillis();
????????}
????????return?REQ_COUNT.incrementAndGet()?<=?QPS;
????}

????public?static?void?main(String[]?args)?throws?InterruptedException?{
????????for?(int?i?=?0;?i?10;?i++)?{
????????????Thread.sleep(250);
????????????LocalTime?now?=?LocalTime.now();
????????????if?(!tryAcquire())?{
????????????????System.out.println(now?+?"?被限流");
????????????}?else?{
????????????????System.out.println(now?+?"?做點什么");
????????????}
????????}
????}
}

運(yùn)行結(jié)果：

20:53:43.038922 做點什么
20:53:43.291435 做點什么
20:53:43.543087 被限流
20:53:43.796666 做點什么
20:53:44.050855 做點什么
20:53:44.303547 被限流
20:53:44.555008 被限流
20:53:44.809083 做點什么
20:53:45.063828 做點什么
20:53:45.314433 被限流

從輸出結(jié)果中可以看到大概每秒操作 3 次，由于限制 QPS 為 2，所以平均會有一次被限流。看起來可以了，不過我們思考一下就會發(fā)現(xiàn)這種簡單的限流方式是有問題的，雖然我們限制了 QPS 為 2，但是當(dāng)遇到時間窗口的臨界突變時，如 1s 中的后 500 ms 和第 2s 的前 500ms 時，雖然是加起來是 1s 時間，卻可以被請求 4 次。

簡單修改測試代碼，可以進(jìn)行驗證：

//?先休眠 400ms，可以更快的到達(dá)時間窗口。
Thread.sleep(400);
for?(int?i?=?0;?i?10;?i++)?{
????Thread.sleep(250);
????if?(!tryAcquire())?{
????????System.out.println("被限流");
????}?else?{
????????System.out.println("做點什么");
????}
}

得到輸出中可以看到連續(xù) 4 次請求，間隔 250 ms 沒有卻被限制。：

20:51:17.395087?做點什么
20:51:17.653114?做點什么
20:51:17.903543?做點什么
20:51:18.154104?被限流
20:51:18.405497?做點什么
20:51:18.655885?做點什么
20:51:18.906177?做點什么
20:51:19.158113?被限流
20:51:19.410512?做點什么
20:51:19.661629?做點什么

3. 滑動窗口算法

我們已經(jīng)知道固定窗口算法的實現(xiàn)方式以及它所存在的問題，而滑動窗口算法是對固定窗口算法的改進(jìn)。既然固定窗口算法在遇到時間窗口的臨界突變時會有問題，那么我們在遇到下一個時間窗口前也調(diào)整時間窗口不就可以了嗎？

下面是滑動窗口的示意圖。

上圖的示例中，每 500ms 滑動一次窗口，可以發(fā)現(xiàn)窗口滑動的間隔越短，時間窗口的臨界突變問題發(fā)生的概率也就越小，不過只要有時間窗口的存在，還是有可能發(fā)生時間窗口的臨界突變問題。

3.1. 代碼實現(xiàn)

下面是基于以上滑動窗口思路實現(xiàn)的簡單的滑動窗口限流工具類。

package?com.wdbyte.rate.limiter;

import?java.time.LocalTime;
import?java.util.concurrent.atomic.AtomicInteger;

/**
?*?滑動窗口限流工具類
?*
?*?@author?https://www.wdbyte.com
?*/
public?class?RateLimiterSlidingWindow?{
????/**
?????*?閾值
?????*/
????private?int?qps?=?2;
????/**
?????*?時間窗口總大小（毫秒）
?????*/
????private?long?windowSize?=?1000;
????/**
?????*?多少個子窗口
?????*/
????private?Integer?windowCount?=?10;
????/**
?????*?窗口列表
?????*/
????private?WindowInfo[]?windowArray?=?new?WindowInfo[windowCount];

????public?RateLimiterSlidingWindow(int?qps)?{
????????this.qps?=?qps;
????????long?currentTimeMillis?=?System.currentTimeMillis();
????????for?(int?i?=?0;?i?????????????windowArray[i]?=?new?WindowInfo(currentTimeMillis,?new?AtomicInteger(0));
????????}
????}

????/**
?????*?1.?計算當(dāng)前時間窗口
?????*?2.?更新當(dāng)前窗口計數(shù)?&?重置過期窗口計數(shù)
?????*?3.?當(dāng)前?QPS?是否超過限制
?????*
?????*?@return
?????*/
????public?synchronized?boolean?tryAcquire()?{
????????long?currentTimeMillis?=?System.currentTimeMillis();
????????//?1.?計算當(dāng)前時間窗口
????????int?currentIndex?=?(int)(currentTimeMillis?%?windowSize?/?(windowSize?/?windowCount));
????????//?2.??更新當(dāng)前窗口計數(shù)?&?重置過期窗口計數(shù)
????????int?sum?=?0;
????????for?(int?i?=?0;?i?????????????WindowInfo?windowInfo?=?windowArray[i];
????????????if?((currentTimeMillis?-?windowInfo.getTime())?>?windowSize)?{
????????????????windowInfo.getNumber().set(0);
????????????????windowInfo.setTime(currentTimeMillis);
????????????}
????????????if?(currentIndex?==?i?&&?windowInfo.getNumber().get()?????????????????windowInfo.getNumber().incrementAndGet();
????????????}
????????????sum?=?sum?+?windowInfo.getNumber().get();
????????}
????????//?3.?當(dāng)前?QPS?是否超過限制
????????return?sum?<=?qps;
????}

????private?class?WindowInfo?{
????????//?窗口開始時間
????????private?Long?time;
????????//?計數(shù)器
????????private?AtomicInteger?number;

????????public?WindowInfo(long?time,?AtomicInteger?number)?{
????????????this.time?=?time;
????????????this.number?=?number;
????????}
????????//?get...set...
????}
}

下面是測試用例，設(shè)置 QPS 為 2,測試次數(shù) 20 次，每次間隔 300 毫秒，預(yù)計成功次數(shù)在 12 次左右。

public?static?void?main(String[]?args)?throws?InterruptedException?{
????int?qps?=?2,?count?=?20,?sleep?=?300,?success?=?count?*?sleep?/?1000?*?qps;
????System.out.println(String.format("當(dāng)前QPS限制為:%d,當(dāng)前測試次數(shù):%d,間隔:%dms,預(yù)計成功次數(shù):%d",?qps,?count,?sleep,?success));
????success?=?0;
????RateLimiterSlidingWindow?myRateLimiter?=?new?RateLimiterSlidingWindow(qps);
????for?(int?i?=?0;?i?????????Thread.sleep(sleep);
????????if?(myRateLimiter.tryAcquire())?{
????????????success++;
????????????if?(success?%?qps?==?0)?{
????????????????System.out.println(LocalTime.now()?+?":?success,?");
????????????}?else?{
????????????????System.out.print(LocalTime.now()?+?":?success,?");
????????????}
????????}?else?{
????????????System.out.println(LocalTime.now()?+?":?fail");
????????}
????}
????System.out.println();
????System.out.println("實際測試成功次數(shù):"?+?success);
}

下面是測試的結(jié)果。

當(dāng)前QPS限制為:2,當(dāng)前測試次數(shù):20,間隔:300ms,預(yù)計成功次數(shù):12
16:04:27.077782:?success,?16:04:27.380715:?success,?
16:04:27.684244:?fail
16:04:27.989579:?success,?16:04:28.293347:?success,?
16:04:28.597658:?fail
16:04:28.901688:?fail
16:04:29.205262:?success,?16:04:29.507117:?success,?
16:04:29.812188:?fail
16:04:30.115316:?fail
16:04:30.420596:?success,?16:04:30.725897:?success,?
16:04:31.028599:?fail
16:04:31.331047:?fail
16:04:31.634127:?success,?16:04:31.939411:?success,?
16:04:32.242380:?fail
16:04:32.547626:?fail
16:04:32.847965:?success,?
實際測試成功次數(shù):11

4. 滑動日志算法

滑動日志算法是實現(xiàn)限流的另一種方法，這種方法比較簡單。基本邏輯就是記錄下所有的請求時間點，新請求到來時先判斷最近指定時間范圍內(nèi)的請求數(shù)量是否超過指定閾值，由此來確定是否達(dá)到限流，這種方式?jīng)]有了時間窗口突變的問題，限流比較準(zhǔn)確，但是因為要記錄下每次請求的時間點，所以占用的內(nèi)存較多。

4.1. 代碼實現(xiàn)

下面是簡單實現(xiàn)的 一個滑動日志算法，因為滑動日志要每次請求單獨存儲一條記錄，可能占用內(nèi)存過多。所以下面這個實現(xiàn)其實不算嚴(yán)謹(jǐn)?shù)幕瑒尤罩荆褚粋€把 1 秒時間切分成 1000 個時間窗口的滑動窗口算法。

package?com.wdbyte.rate.limiter;

import?java.time.LocalTime;
import?java.util.HashSet;
import?java.util.Set;
import?java.util.TreeMap;

/**
?*?滑動日志方式限流
?*?設(shè)置?QPS?為?2.
?*
?*?@author?https://www.wdbyte.com
?*/
public?class?RateLimiterSildingLog?{

????/**
?????*?閾值
?????*/
????private?Integer?qps?=?2;
????/**
?????*?記錄請求的時間戳,和數(shù)量
?????*/
????private?TreeMap?treeMap?=?new?TreeMap<>();

????/**
?????*?清理請求記錄間隔,?60?秒
?????*/
????private?long?claerTime?=?60?*?1000;

????public?RateLimiterSildingLog(Integer?qps)?{
????????this.qps?=?qps;
????}

????public?synchronized?boolean?tryAcquire()?{
????????long?now?=?System.currentTimeMillis();
????????//?清理過期的數(shù)據(jù)老數(shù)據(jù)，最長?60?秒清理一次
????????if?(!treeMap.isEmpty()?&&?(treeMap.firstKey()?-?now)?>?claerTime)?{
????????????Set?keySet?=?new?HashSet<>(treeMap.subMap(0L,?now?-?1000).keySet());
????????????for?(Long?key?:?keySet)?{
????????????????treeMap.remove(key);
????????????}
????????}
????????//?計算當(dāng)前請求次數(shù)
????????int?sum?=?0;
????????for?(Long?value?:?treeMap.subMap(now?-?1000,?now).values())?{
????????????sum?+=?value;
????????}
????????//?超過QPS限制，直接返回?false
????????if?(sum?+?1?>?qps)?{
????????????return?false;
????????}
????????//?記錄本次請求
????????if?(treeMap.containsKey(now))?{
????????????treeMap.compute(now,?(k,?v)?->?v?+?1);
????????}?else?{
????????????treeMap.put(now,?1L);
????????}
????????return?sum?<=?qps;
????}

????public?static?void?main(String[]?args)?throws?InterruptedException?{
????????RateLimiterSildingLog?rateLimiterSildingLog?=?new?RateLimiterSildingLog(3);
????????for?(int?i?=?0;?i?10;?i++)?{
????????????Thread.sleep(250);
????????????LocalTime?now?=?LocalTime.now();
????????????if?(rateLimiterSildingLog.tryAcquire())?{
????????????????System.out.println(now?+?"?做點什么");
????????????}?else?{
????????????????System.out.println(now?+?"?被限流");
????????????}
????????}
????}
}

代碼中把閾值 QPS 設(shè)定為 3，運(yùn)行可以得到如下日志：

20:51:17.395087?做點什么
20:51:17.653114?做點什么
20:51:17.903543?做點什么
20:51:18.154104?被限流
20:51:18.405497?做點什么
20:51:18.655885?做點什么
20:51:18.906177?做點什么
20:51:19.158113?被限流
20:51:19.410512?做點什么
20:51:19.661629?做點什么

5. 漏桶算法

漏桶算法中的漏桶是一個形象的比喻，這里可以用生產(chǎn)者消費(fèi)者模式進(jìn)行說明，請求是一個生產(chǎn)者，每一個請求都如一滴水，請求到來后放到一個隊列（漏桶）中，而桶底有一個孔，不斷的漏出水滴，就如消費(fèi)者不斷的在消費(fèi)隊列中的內(nèi)容，消費(fèi)的速率（漏出的速度）等于限流閾值。即假如 QPS ?為 2，則每 1s / 2= 500ms 消費(fèi)一次。漏桶的桶有大小，就如隊列的容量，當(dāng)請求堆積超過指定容量時，會觸發(fā)拒絕策略。

下面是漏桶算法的示意圖。

由介紹可以知道，漏桶模式中的消費(fèi)處理總是能以恒定的速度進(jìn)行，可以很好的保護(hù)自身系統(tǒng)不被突如其來的流量沖垮；但是這也是漏桶模式的缺點，假設(shè) QPS 為 2，同時 2 個請求進(jìn)來，2 個請求并不能同時進(jìn)行處理響應(yīng)，因為每 1s / 2= 500ms 只能處理一個請求。

6. 令牌桶算法

令牌桶算法同樣是實現(xiàn)限流是一種常見的思路，最為常用的 Google 的 Java 開發(fā)工具包 Guava 中的限流工具類 RateLimiter 就是令牌桶的一個實現(xiàn)。令牌桶的實現(xiàn)思路類似于生產(chǎn)者和消費(fèi)之間的關(guān)系。

系統(tǒng)服務(wù)作為生產(chǎn)者，按照指定頻率向桶（容器）中添加令牌，如 QPS 為 2，每 500ms 向桶中添加一個令牌，如果桶中令牌數(shù)量達(dá)到閾值，則不再添加。

請求執(zhí)行作為消費(fèi)者，每個請求都需要去桶中拿取一個令牌，取到令牌則繼續(xù)執(zhí)行；如果桶中無令牌可取，就觸發(fā)拒絕策略，可以是超時等待，也可以是直接拒絕本次請求，由此達(dá)到限流目的。

下面是令牌桶限流算法示意圖。

思考令牌桶的實現(xiàn)可以以下特點。

1. 1s / 閾值（QPS） ?= 令牌添加時間間隔。
2. 桶的容量等于限流的閾值，令牌數(shù)量達(dá)到閾值時，不再添加。
3. 可以適應(yīng)流量突發(fā)，N 個請求到來只需要從桶中獲取 N 個令牌就可以繼續(xù)處理。
4. 有啟動過程，令牌桶啟動時桶中無令牌，然后按照令牌添加時間間隔添加令牌，若啟動時就有閾值數(shù)量的請求過來，會因為桶中沒有足夠的令牌而觸發(fā)拒絕策略，不過如 RateLimiter 限流工具已經(jīng)優(yōu)化了這類問題。

6.1. 代碼實現(xiàn)

Google 的 Java 開發(fā)工具包 Guava 中的限流工具類 RateLimiter 就是令牌桶的一個實現(xiàn)，日常開發(fā)中我們也不會手動實現(xiàn)了，這里直接使用 RateLimiter 進(jìn)行測試。

引入依賴：

<exclusion>
???<groupId>com.google.guavagroupId>
????<artifactId>guavaartifactId>
???<version>31.0.1-jreversion>
exclusion>

RateLimiter 限流體驗：

//?qps?2
RateLimiter?rateLimiter?=?RateLimiter.create(2);
for?(int?i?=?0;?i?10;?i++)?{
????String?time?=?LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
????System.out.println(time?+?":"?+?rateLimiter.tryAcquire());
????Thread.sleep(250);
}

代碼中限制 QPS 為 2，也就是每隔 500ms 生成一個令牌，但是程序每隔 250ms 獲取一次令牌，所以兩次獲取中只有一次會成功。

17:19:06.797557:true
17:19:07.061419:false
17:19:07.316283:true
17:19:07.566746:false
17:19:07.817035:true
17:19:08.072483:false
17:19:08.326347:true
17:19:08.577661:false
17:19:08.830252:true
17:19:09.085327:false

6.2. 思考

雖然演示了 Google Guava 工具包中的 RateLimiter 的實現(xiàn)，但是我們需要思考一個問題，就是令牌的添加方式，如果按照指定間隔添加令牌，那么需要開一個線程去定時添加，如果有很多個接口很多個 RateLimiter 實例，線程數(shù)會隨之增加，這顯然不是一個好的辦法。顯然 Google 也考慮到了這個問題，在 RateLimiter 中，是在每次令牌獲取時才進(jìn)行計算令牌是否足夠的。它通過存儲的下一個令牌生成的時間，和當(dāng)前獲取令牌的時間差，再結(jié)合閾值，去計算令牌是否足夠，同時再記錄下一個令牌的生成時間以便下一次調(diào)用。

下面是 Guava 中 RateLimiter 類的子類 SmoothRateLimiter 的 resync() 方法的代碼分析，可以看到其中的令牌計算邏輯。

void?resync(long?nowMicros)?{?//?當(dāng)前微秒時間
????//?當(dāng)前時間是否大于下一個令牌生成時間
????if?(nowMicros?>?this.nextFreeTicketMicros)?{?
???????//?可生成的令牌數(shù) newPermits =?（當(dāng)前時間?-?下一個令牌生成時間）/?令牌生成時間間隔。
???????//?如果?QPS?為2，這里的?coolDownIntervalMicros?就是?500000.0?微秒(500ms)
????????double?newPermits?=?(double)(nowMicros?-?this.nextFreeTicketMicros)?/?this.coolDownIntervalMicros();
????//?更新令牌庫存 storedPermits。
???????this.storedPermits?=?Math.min(this.maxPermits,?this.storedPermits?+?newPermits);
????//?更新下一個令牌生成時間?nextFreeTicketMicros
???????this.nextFreeTicketMicros?=?nowMicros;
????}
}

7. Redis 分布式限流

Redis 是一個開源的內(nèi)存數(shù)據(jù)庫，可以用來作為數(shù)據(jù)庫、緩存、消息中間件等。Redis 是單線程的，又在內(nèi)存中操作，所以速度極快，得益于 Redis 的各種特性，所以使用 Redis 實現(xiàn)一個限流工具是十分方便的。

下面的演示都基于Spring Boot 項目，并需要以下依賴。

    org.springframework.boot
    spring-boot-starter-data-redis

配置 Redis 信息。

spring:
??redis:
????database:?0
????password:?
????port:?6379
????host:?127.0.0.1
????lettuce:
??????shutdown-timeout:?100ms
??????pool:
????????min-idle:?5
????????max-idle:?10
????????max-active:?8
????????max-wait:?1ms

7.1. 固定窗口限流

Redis 中的固定窗口限流是使用 incr 命令實現(xiàn)的，incr 命令通常用來自增計數(shù)；如果我們使用時間戳信息作為 key，自然就可以統(tǒng)計每秒的請求量了，以此達(dá)到限流目的。

這里有兩點要注意。

1. 對于不存在的 key，第一次新增時，value 始終為 1。
2. INCR 和 EXPIRE 命令操作應(yīng)該在一個原子操作中提交，以保證每個 key 都正確設(shè)置了過期時間，不然會有 key 值無法自動刪除而導(dǎo)致的內(nèi)存溢出。

由于 Redis 中實現(xiàn)事務(wù)的復(fù)雜性，所以這里直接只用 lua 腳本來實現(xiàn)原子操作。下面是 lua 腳本內(nèi)容。

local?count?=?redis.call("incr",KEYS[1])
if?count?==?1?then
??redis.call('expire',KEYS[1],ARGV[2])
end
if?count?>?tonumber(ARGV[1])?then
??return?0
end
return?1

下面是使用 Spring Boot 中 RedisTemplate 來實現(xiàn)的 lua 腳本調(diào)用測試代碼。

/**
?*?@author?https://www.wdbyte.com
?*/
@SpringBootTest
class?RedisLuaLimiterByIncr?{
????private?static?String?KEY_PREFIX?=?"limiter_";
????private?static?String?QPS?=?"4";
????private?static?String?EXPIRE_TIME?=?"1";

????@Autowired
????private?StringRedisTemplate?stringRedisTemplate;

????@Test
????public?void?redisLuaLimiterTests()?throws?InterruptedException,?IOException?{
????????for?(int?i?=?0;?i?15;?i++)?{
????????????Thread.sleep(200);
????????????System.out.println(LocalTime.now()?+?"?"?+?acquire("user1"));
????????}
????}

????/**
?????*?計數(shù)器限流
?????*
?????*?@param?key
?????*?@return
?????*/
????public?boolean?acquire(String?key)?{
????????//?當(dāng)前秒數(shù)作為?key
????????key?=?KEY_PREFIX?+?key?+?System.currentTimeMillis()?/?1000;
????????DefaultRedisScript?redisScript?=?new?DefaultRedisScript<>();
????????redisScript.setResultType(Long.class);
????????//lua文件存放在resources目錄下
????????redisScript.setScriptSource(new?ResourceScriptSource(new?ClassPathResource("limiter.lua")));
????????return?stringRedisTemplate.execute(redisScript,?Arrays.asList(key),?QPS,?EXPIRE_TIME)?==?1;
????}
}

代碼中雖然限制了 QPS 為 4，但是因為這種限流實現(xiàn)是把毫秒時間戳作為 key 的，所以會有臨界窗口突變的問題，下面是運(yùn)行結(jié)果，可以看到因為時間窗口的變化，導(dǎo)致了 QPS 超過了限制值 4。

17:38:23.122044?true
17:38:23.695124?true
17:38:23.903220?true
#?此處有時間窗口變化，所以下面繼續(xù)?true
17:38:24.106206?true
17:38:24.313458?true
17:38:24.519431?true
17:38:24.724446?true
17:38:24.932387?false
17:38:25.137912?true
17:38:25.355595?true
17:38:25.558219?true
17:38:25.765801?true
17:38:25.969426?false
17:38:26.176220?true
17:38:26.381918?true

7.3. 滑動窗口限流

通過對上面的基于 incr 命令實現(xiàn)的 Redis 限流方式的測試，我們已經(jīng)發(fā)現(xiàn)了固定窗口限流所帶來的問題，在這篇文章的第三部分已經(jīng)介紹了滑動窗口限流的優(yōu)勢，它可以大幅度降低因為窗口臨界突變帶來的問題，那么如何使用 Redis 來實現(xiàn)滑動窗口限流呢？

這里主要使用 ZSET 有序集合來實現(xiàn)滑動窗口限流，ZSET 集合有下面幾個特點：

1. ZSET 集合中的 ?key 值可以自動排序。
2. ZSET 集合中的 value 不能有重復(fù)值。
3. ZSET 集合可以方便的使用 ZCARD 命令獲取元素個數(shù)。
4. ZSET 集合可以方便的使用 ZREMRANGEBYLEX 命令移除指定范圍的 key 值。

基于上面的四點特性，可以編寫出基于 ZSET 的滑動窗口限流 lua 腳本。

--KEYS[1]:?限流?key
--ARGV[1]:?時間戳?-?時間窗口
--ARGV[2]:?當(dāng)前時間戳（作為score）
--ARGV[3]:?閾值
--ARGV[4]:?score?對應(yīng)的唯一value
--?1.?移除時間窗口之前的數(shù)據(jù)
redis.call('zremrangeByScore',?KEYS[1],?0,?ARGV[1])
--?2.?統(tǒng)計當(dāng)前元素數(shù)量
local?res?=?redis.call('zcard',?KEYS[1])
--?3.?是否超過閾值
if?(res?==?nil)?or?(res?tonumber(ARGV[3]))?then
????redis.call('zadd',?KEYS[1],?ARGV[2],?ARGV[4])
????return?1
else
????return?0
end

下面是使用 Spring Boot 中 RedisTemplate 來實現(xiàn)的 lua 腳本調(diào)用測試代碼。

@SpringBootTest
class?RedisLuaLimiterByZset?{

????private?String?KEY_PREFIX?=?"limiter_";
????private?String?QPS?=?"4";

????@Autowired
????private?StringRedisTemplate?stringRedisTemplate;

????@Test
????public?void?redisLuaLimiterTests()?throws?InterruptedException,?IOException?{
????????for?(int?i?=?0;?i?15;?i++)?{
????????????Thread.sleep(200);
????????????System.out.println(LocalTime.now()?+?"?"?+?acquire("user1"));
????????}
????}

????/**
?????*?計數(shù)器限流
?????*
?????*?@param?key
?????*?@return
?????*/
????public?boolean?acquire(String?key)?{
????????long?now?=?System.currentTimeMillis();
????????key?=?KEY_PREFIX?+?key;
????????String?oldest?=?String.valueOf(now?-?1_000);
????????String?score?=?String.valueOf(now);
????????String?scoreValue?=?score;
????????DefaultRedisScript?redisScript?=?new?DefaultRedisScript<>();
????????redisScript.setResultType(Long.class);
????????//lua文件存放在resources目錄下
????????redisScript.setScriptSource(new?ResourceScriptSource(new?ClassPathResource("limiter2.lua")));
????????return?stringRedisTemplate.execute(redisScript,?Arrays.asList(key),?oldest,?score,?QPS,?scoreValue)?==?1;
????}
}

代碼中限制 QPS 為 4，運(yùn)行結(jié)果信息與之一致。

17:36:37.150370?true
17:36:37.716341?true
17:36:37.922577?true
17:36:38.127497?true
17:36:38.335879?true
17:36:38.539225?false
17:36:38.745903?true
17:36:38.952491?true
17:36:39.159497?true
17:36:39.365239?true
17:36:39.570572?false
17:36:39.776635?true
17:36:39.982022?true
17:36:40.185614?true
17:36:40.389469?true

這里介紹了 Redis 實現(xiàn)限流的兩種方式，當(dāng)然使用 Redis 也可以實現(xiàn)漏桶和令牌桶兩種限流算法，這里就不做演示了，感興趣的可以自己研究下。

8. 總結(jié)

這篇文章介紹實現(xiàn)限流的幾種方式，主要是窗口算法和桶算法，兩者各有優(yōu)勢。

• 窗口算法實現(xiàn)簡單，邏輯清晰，可以很直觀的得到當(dāng)前的 QPS 情況，但是會有時間窗口的臨界突變問題，而且不像桶一樣有隊列可以緩沖。
• 桶算法雖然稍微復(fù)雜，不好統(tǒng)計 QPS 情況，但是桶算法也有優(yōu)勢所在。
• 漏桶模式消費(fèi)速率恒定，可以很好的保護(hù)自身系統(tǒng)，可以對流量進(jìn)行整形，但是面對突發(fā)流量不能快速響應(yīng)。
• 令牌桶模式可以面對突發(fā)流量，但是啟動時會有緩慢加速的過程，不過常見的開源工具中已經(jīng)對此優(yōu)化。

單機(jī)限流與分布式限流

上面演示的基于代碼形式的窗口算法和桶算法限流都適用于單機(jī)限流，如果需要分布式限流可以結(jié)合注冊中心、負(fù)載均衡計算每個服務(wù)的限流閾值，但這樣會降低一定精度，如果對精度要求不是太高，可以使用。

而 Redis 的限流，由于 Redis 的單機(jī)性，本身就可以用于分布式限流。使用 Redis 可以實現(xiàn)各種可以用于限流算法，如果覺得麻煩也可以使用開源工具如 redisson，已經(jīng)封裝了基于 Redis 的限流。

其他限流工具

文中已經(jīng)提到了 Guava 的限流工具包，不過它畢竟是單機(jī)的，開源社區(qū)中也有很多分布式限流工具，如阿里開源的 Sentinel 就是不錯的工具，Sentinel 以流量為切入點，從流量控制、熔斷降級、系統(tǒng)負(fù)載保護(hù)等多個維度保護(hù)服務(wù)的穩(wěn)定性。

一如既往，文章中的代碼存放在：github.com/niumoo/JavaNotes

參考

Redis INCR:https://redis.io/commands/incr

Rate Limiting Wikipedia：https://en.wikipedia.org/wiki/Rate_limiting

SpringBoot Redis:https://www.cnblogs.com/lenve/p/10965667.html

··········? END? ··············