回想我們?cè)阢y行和政府機(jī)關(guān)去辦事時(shí)卷仑， 都會(huì)有一個(gè)排隊(duì)機(jī)峻村， 先取一個(gè)號(hào)， 然后等待叫號(hào)锡凝， 辦事窗口多粘昨， 號(hào)就叫得快， 辦事窗口少窜锯， 號(hào)就叫得慢张肾， 排隊(duì)機(jī)是一個(gè)了不起的發(fā)明， 這里有許多值得我們?cè)诰幊虝r(shí)借鑒的東西锚扎。
- 它其實(shí)應(yīng)用了 leader/follower 的并發(fā)模式吞瞪， 每個(gè)辦事窗口就是一個(gè)工作線程， 由排號(hào)機(jī)這個(gè) leader 來(lái)分配工作
- 它其實(shí)應(yīng)用了限流模式驾孔， 用排號(hào)限制了任務(wù)的擁塞芍秆，再多的人來(lái)銀行辦事， 銀行也能 hold 得住翠勉， 大不了廣而告之妖啥， 今天的號(hào)發(fā)完了， 明天請(qǐng)?jiān)?/li>
- 它其實(shí)是一個(gè)消息隊(duì)列系統(tǒng)
- 它其實(shí)是一個(gè)事件驅(qū)動(dòng)系統(tǒng)
- 它其實(shí)就是一個(gè)排隊(duì)機(jī)
閑言少敘对碌， 書(shū)歸正傳荆虱， 我們來(lái)重點(diǎn)講講限流這回事， 任何系統(tǒng)都有容量的限制朽们， 為了使我們的服務(wù)保持高可用性克伊， 我們必須對(duì)系統(tǒng)進(jìn)行限流， 也稱速率限制 Rate Limiting
限流和流量控制也還有點(diǎn)區(qū)別华坦，在傳輸層協(xié)議層面上就已經(jīng)做了一些流量控制愿吹， TCP 通過(guò)可變大小的滑動(dòng)窗口來(lái)進(jìn)行數(shù)據(jù)傳輸?shù)牧髁靠刂疲?jiǎn)單來(lái)說(shuō)惜姐， 發(fā)送方有一個(gè)滑動(dòng)窗口犁跪， 大小為10, 也就是說(shuō)發(fā)送10個(gè)字節(jié)之后才等待接收方的響應(yīng)， 接收方在接收確認(rèn)消息中包含一個(gè) window advertisement 窗口建議告之發(fā)送方 -- 作為接收方準(zhǔn)備好接收多少字節(jié)的數(shù)據(jù)歹袁， 這個(gè)值如果比較大坷衍，那么發(fā)送方的滑動(dòng)窗口可以增大 ， 可以快點(diǎn)發(fā)送數(shù)據(jù)条舔， 因?yàn)榻邮辗降奶幚硇屎芨撸?反之枫耳， 則減小滑動(dòng)窗口大小， 這樣就減慢了發(fā)送速率孟抗， 當(dāng)滑動(dòng)窗口大小為1時(shí)迁杨， 則發(fā)送每個(gè)消息都要等待確認(rèn)消息收到后才發(fā)送下一個(gè)
大多數(shù)的消息隊(duì)列系統(tǒng)也用到了 Flow Control 钻心， 當(dāng)生產(chǎn)者過(guò)快地發(fā)送消息， 而消費(fèi)者沒(méi)法及時(shí)處理時(shí)铅协，并返回一個(gè)異常消息捷沸，告之生產(chǎn)者搞慢點(diǎn) -- “馬兒你慢點(diǎn)走慢點(diǎn)走吔”
而這里講的限流是指速率控制， 服務(wù)器端對(duì)客戶端的請(qǐng)求進(jìn)行監(jiān)控狐史，當(dāng)發(fā)覺(jué)某個(gè)客戶端發(fā)送了過(guò)多或過(guò)快的請(qǐng)求就會(huì)做出限制痒给， 根據(jù)預(yù)先制定的策略針對(duì)某個(gè)客戶端的IP, 帳戶或類(lèi)型進(jìn)行限流， 從而保證了對(duì)大多數(shù)正常請(qǐng)求的服務(wù)不受影響, 防止拒絕服務(wù) DoS (denial of service) 和分布式拒絕服務(wù) DDoS (distributed denial of service) , DoS 是很常見(jiàn)的網(wǎng)絡(luò)攻擊方式, 限流或者說(shuō)速率控制 Rate Limiting 是行之有效的應(yīng)對(duì)手段.
限流可以是比較粗放式的, 只根據(jù)每秒請(qǐng)求數(shù)的閾值來(lái)進(jìn)行控制, 超過(guò) QPS/TPS 上限的請(qǐng)求一律拒絕掉, 這種方式 有效, 但是不能精準(zhǔn)打擊那些攻擊者, 反而會(huì)誤傷無(wú)辜.
我們可以縮小限制范圍, 按照如下三個(gè)級(jí)別來(lái)作區(qū)別對(duì)待
- 源地址層面 Source Address Level
我們從 HTTP 請(qǐng)求中都可以得到 TCP 頭中的 source address, 如果來(lái)自一個(gè)源地址的請(qǐng)求過(guò)多過(guò)快, 可以將它超過(guò)閾值的請(qǐng)求拒絕掉, 其他來(lái)源的合法請(qǐng)求不做限制.
這里要著重注意一點(diǎn), 用戶的請(qǐng)求一般不會(huì)直接到達(dá)服務(wù)器, 而是會(huì)經(jīng)過(guò)一個(gè)負(fù)載均衡器 Load Balancer 分流后到達(dá)服務(wù)器, 所以這個(gè)源地址很可能就變成了 Load Balancer 的地址, 所以我們最好先檢查一下在 HTTP 頭域中是否有 X-Forwarded-For , 這是由負(fù)載均衡器所添加的來(lái)自客戶端的真實(shí)地址, 意謂轉(zhuǎn)發(fā)自何處.
眾多的 HTTP 代理服務(wù)器, 也會(huì)添加這個(gè) HTTP 頭域, 如果有這個(gè)頭域, 那么就應(yīng)該以它作為客戶端的IP 加以區(qū)分, 無(wú)論是硬件的 F5, Netscalar, 還是軟件版本的 Nginx, HAProxy 都支持這一選項(xiàng).
- 用戶層面 User level
微服務(wù)對(duì)外提供的 API , 首先需要通過(guò)認(rèn)證 Authentication 和授權(quán) Authorization, 一旦認(rèn)證和授權(quán)通過(guò), 我們就能得知這個(gè)請(qǐng)求所代表的用戶信息, 針對(duì)這個(gè)信息, 可以做基于用戶分組的限流.
例如, 我們?cè)诜?wù)器簽發(fā)的 token 中包含用戶信息: userId, orgId, 然后就可以針對(duì) userId 或 orgId 做單獨(dú)的計(jì)數(shù), 如果在特定時(shí)間單位中超過(guò)最大數(shù)量閾值, 則拒絕此特定用戶或組織的請(qǐng)求. 實(shí)際應(yīng)用中就可以在 JWT(Json Web Token) 中添加自定義字段來(lái)表示用戶, 組織及應(yīng)用程序的標(biāo)識(shí)信息.
- 應(yīng)用程序?qū)用? Application Level
類(lèi)似用戶層面的限流, 一旦我們可以辨別出應(yīng)用程序的標(biāo)識(shí), 就可以針對(duì)特定應(yīng)用程序的請(qǐng)求進(jìn)行計(jì)數(shù), 按照下面介紹的限流算法來(lái)進(jìn)行速率控制.
限流算法
Leaky Bucket 漏桶
就象我們生活中常見(jiàn)的漏斗, 從油桶往油瓶里倒油, 沒(méi)有漏斗, 除非是賣(mài)油翁那樣的專(zhuān)家, 多數(shù)情況下油都會(huì)跑冒滴漏.
漏桶是類(lèi)似的東西, 海量請(qǐng)求撲面而來(lái), 可能瞬時(shí)就會(huì)把服務(wù)壓垮, 而漏桶就可以用來(lái)限流削峰.
漏桶是總?cè)萘渴遣蛔兊? 水滴(請(qǐng)求) 以任意速率流入, 但總是以恒定速率流出, 如果請(qǐng)求來(lái)得太多太快, 桶的容量就會(huì)撐滿, 后續(xù)的請(qǐng)求就會(huì)被拒絕, 也就是說(shuō)當(dāng)一個(gè)請(qǐng)求到來(lái), 就流一滴水進(jìn)桶里,如果可以放入, 則處理此請(qǐng)求, 否則漏桶已滿, 則拒絕此請(qǐng)求, 直到桶中水滴不再滿時(shí)
Token Bucket 令牌桶
令牌桶與上而的漏桶異曲同工, 只不過(guò)它不是以固定速率流出, 而是以固定速率放入令牌到令牌桶中, 請(qǐng)求到來(lái)時(shí)從令牌桶中領(lǐng)取一個(gè)令牌才可繼續(xù)處理服務(wù), 如果取不到令牌, 則拒絕此請(qǐng)求
Fixed Window 固定窗口
固定窗口算法, 也就是用一個(gè)固定的時(shí)間窗口來(lái)跟蹤速率, 每一個(gè)請(qǐng)求會(huì)增加這個(gè)窗口中的計(jì)數(shù)器, 請(qǐng)求來(lái)了加1, 處理完成就會(huì)減1, 如果這個(gè)計(jì)數(shù)器超過(guò)了閾值, 后續(xù)的請(qǐng)求就會(huì)丟棄掉.
比如60秒的窗口設(shè)置閾值為1200, 12:00 ~ 12:01 就是一個(gè)窗口, 在這個(gè)窗口期中的請(qǐng)求數(shù)超過(guò)了1200, 再進(jìn)來(lái)的請(qǐng)求就會(huì)丟棄掉.
Sliding Log 滑動(dòng)日志
以時(shí)間戳為 key 將請(qǐng)求日志保存在一張表中, 每個(gè)請(qǐng)求都會(huì)在這張表中添加一條日志, 日志的生存周期(TTL - Time To Live)有限, 過(guò)期的日志會(huì)被刪除, 如果表中所存儲(chǔ)的日志數(shù)已經(jīng)達(dá)到了閾值, 后續(xù)的新請(qǐng)求就會(huì)丟棄掉
Sliding Window 滑動(dòng)窗口
這是一種改進(jìn)算法, 綜合了固定窗口和滑動(dòng)日志兩種方法的優(yōu)點(diǎn)骏全，它結(jié)合了固定窗口算法的低處理成本和滑動(dòng)日志的改進(jìn)邊界條件, 將當(dāng)前時(shí)間窗口與過(guò)去時(shí)間窗口綜合考慮苍柏。 與固定窗口算法一樣，根據(jù)請(qǐng)求更改每個(gè)固定窗口的計(jì)數(shù)器姜贡。 接下來(lái)序仙，再根據(jù)當(dāng)前時(shí)間戳計(jì)算出當(dāng)前窗口的加權(quán)值, 以及上一個(gè)窗口的請(qǐng)求率的加權(quán)值，以平滑流量突發(fā)鲁豪。 例如潘悼，如果當(dāng)前窗口是25％，那么我們將前一個(gè)窗口的計(jì)數(shù)加權(quán)75％爬橡。
限流級(jí)別
其實(shí)就是計(jì)數(shù)器涵蓋的范圍, 通過(guò)實(shí)例級(jí)別也就夠了, 單個(gè)實(shí)例超過(guò)流量了, 由于前面都有一個(gè)負(fù)載均衡器存在, 其他實(shí)例大概率也會(huì)過(guò)載.
大致我們可以分為在下四個(gè)級(jí)別
實(shí)例級(jí)別 Instance
服務(wù)器級(jí)別 Servrer
集群級(jí)別 Pool/Cluster
數(shù)據(jù)中心 Data center
限流范圍
具體到限制范圍, 根據(jù)上述的三個(gè)層面, 源地址, 用戶, 應(yīng)用程序, 還可以加上微服務(wù)自身所提供的不同端點(diǎn)來(lái)劃分范圍
- 全部端點(diǎn) all endpoints/特定端點(diǎn) specified endpoint
- 全部源地址/特定源地址
- 全部用戶/特定用戶
- 全部應(yīng)用程序/特定應(yīng)用程序
限流策略
最簡(jiǎn)單的策略當(dāng)然是直接拒絕, 簡(jiǎn)單粗暴有效, 但是如果想做得比較平滑優(yōu)雅, 可以部分拒絕, 逐步收窄, 對(duì)于那些十分重要的大客戶, 可沒(méi)有額外資源可以調(diào)度的情況, 甚至可以設(shè)置為永不拒絕策略
- 全部拒絕
- 部分拒絕: 在指定時(shí)間間隔內(nèi)允許若干個(gè)請(qǐng)求 QPS, QPM, QPH
- 總不拒絕: 對(duì)于某些非常重要的客戶, 總是允許他們的請(qǐng)求, 直至系統(tǒng)資源耗盡
例如在 HTTP 請(qǐng)求拒絕的時(shí)候我們可以用響應(yīng)碼 "429 Too Many Requests", 還可加上一個(gè) Retry-After 來(lái)建議用戶多長(zhǎng)時(shí)間以后再重試.
限流度量及觸發(fā)條件
根據(jù)上述算法, 關(guān)鍵的度量指標(biāo)就是計(jì)數(shù)器
- 漏桶中的水滴數(shù)是否已經(jīng)達(dá)到閾值
- 令牌桶中的令牌數(shù)是否已經(jīng)領(lǐng)光
- 固定窗口或滑動(dòng)窗口中的計(jì)數(shù)器是否已經(jīng)達(dá)到閾值
- 滑動(dòng)日志中所存儲(chǔ)的日志數(shù)是否已經(jīng)達(dá)到閾值
當(dāng)然可以設(shè)置更加細(xì)致的匹配和分組條件：
例如
- API 端點(diǎn)信息: url, responseCode, header, method, param
- 用戶信息: userId, orgId
- ip 地址信息: source_address, x-forwarded-for
觸發(fā)條件一般是單位時(shí)間內(nèi)的最大請(qǐng)求數(shù), 例如：
- 單位時(shí)間內(nèi)的請(qǐng)求數(shù) request count per interval
- 單位時(shí)間內(nèi)的錯(cuò)誤碼次數(shù) error code count per interval
- 單位時(shí)間內(nèi)的并發(fā)請(qǐng)求數(shù) concurrent request per interval
實(shí)例演示
對(duì)于實(shí)例級(jí)別的限流, 在內(nèi)存中維護(hù)一個(gè)漏桶, 令牌桶, 或者計(jì)數(shù)器就好了
內(nèi)存計(jì)數(shù)器
public interface RateLimiter {
boolean allow();
}
package com.github.walterfan.util.ratelimit;
import com.google.common.util.concurrent.Uninterruptibles;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
public class FixedWindow implements RateLimiter {
private final ConcurrentMap<Long, AtomicInteger> windows = new ConcurrentHashMap<>();
private int maxRequestsPerSecond;
private int windowSizeInMs;
public FixedWindow(int maxReqPerSec, int windowSizeInMs) {
this.maxRequestsPerSecond = maxReqPerSec;
this.windowSizeInMs = windowSizeInMs;
}
@Override
public boolean allow() {
long windowKey = System.currentTimeMillis() / windowSizeInMs;
windows.putIfAbsent(windowKey, new AtomicInteger(0));
//log.debug("counter of {} --> {}", windowKey, windows.get(windowKey));
return windows.get(windowKey).incrementAndGet() <= maxRequestsPerSecond;
}
public String toString() {
StringBuilder sb = new StringBuilder("");
for(Map.Entry<Long, AtomicInteger> entry: windows.entrySet()) {
sb.append(entry.getKey());
sb.append(" --> ");
sb.append(entry.getValue());
sb.append("\n");
}
return sb.toString();
}
public static void main(String[] args) {
FixedWindow fixedWindow = new FixedWindow(10, 1000);
for(int i=0;i<20;i++) {
boolean ret = fixedWindow.allow();
Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
if(!ret)
log.info("{}, ret={}", i, ret);
}
log.info(fixedWindow.toString());
}
}
}
執(zhí)行結(jié)果如下
10, ret=false
11, ret=false
12, ret=false
13, ret=false
14, ret=false
15, ret=false
1558961918 --> 4
1558961917 --> 16
Guava Ratelimiter
Guava 也提供了一個(gè) Rate Limiter 的簡(jiǎn)單實(shí)現(xiàn)
使用方法
final RateLimiter rateLimiter = RateLimiter.create(2.0); // 允許每秒2.0 個(gè)請(qǐng)求 Call Per Second
void submitTasks(List<Runnable> tasks, Executor executor) {
for (Runnable task : tasks) {
// 這里會(huì)阻塞直至請(qǐng)求數(shù)不再達(dá)到閾值 2 CPS,
// 如果不想阻塞, 可使用 tryAcquire(int permits, long timeout, TimeUnit unit)
rateLimiter.acquire();
executor.execute(task);
}
}
Redis 計(jì)數(shù)器
對(duì)于集群級(jí)別的限流, 可以利用 Redis 來(lái)存儲(chǔ)計(jì)算器, 比如我們想對(duì)某一個(gè)API 進(jìn)行限流, 閾值為 100 CPS
在Redis 存儲(chǔ)一張哈希表, key 名為 counter_<api_endpoint_name>_cps:timestamp 為當(dāng)前時(shí)間 System.currentTimeMillis() 除以1000
例如
set counter_check_health_cps:1558962905 1
"OK"
INCRBY counter_check_health_cps:1558962905 1
2
Zuul Route Filter
Spring Cloud 的開(kāi)源網(wǎng)關(guān)項(xiàng)目 Zuul , 它基于過(guò)濾器模式提供若干過(guò)濾器的實(shí)現(xiàn), 對(duì)于 Rate Limit 的也有一個(gè)開(kāi)源的實(shí)現(xiàn)
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-zuul</artifactId>
</dependency>
<dependency>
<groupId>com.marcosbarbero.cloud</groupId>
<artifactId>spring-cloud-zuul-ratelimit</artifactId>
<version>2.2.0.RELEASE</version>
</dependency>
對(duì)于以下 /potato/health API
@Controller
@RequestMapping("/potato")
public class GreetingController {
@GetMapping("/health")
public ResponseEntity<String> getSimple() {
return ResponseEntity.ok("OKOKOK");
}
}
可以設(shè)置 Zuul 針對(duì) CheckHealth的速率控制為 5 CPS(Call Per Second)
zuul:
routes:
checkHealth:
path: /potato/health
url: forward:/
ratelimit:
enabled: true
repository: JPA
policy-list:
checkHealth:
- limit: 5
refresh-interval: 60
type:
- origin
strip-prefix: true
