前記

在微服務(wù)中， 雖然服務(wù)間的調(diào)用都是可信的， 但是服務(wù)端也需要堤防一些流量， 防止被意外的流量擊垮， 而通過(guò)限流可以防止問(wèn)題的發(fā)生。此外， 使用不同的限流規(guī)則還能根據(jù)系統(tǒng)間不同服務(wù)的請(qǐng)求進(jìn)行限制， 解決某個(gè)函數(shù)被頻繁調(diào)用而拖垮整個(gè)系統(tǒng)的問(wèn)題。

NOTE: 雖然本文是在編寫(xiě)RPC框架有感而發(fā)， 但是也適用于常見(jiàn)的Web服務(wù)等有流量進(jìn)出的場(chǎng)景。

最新修訂見(jiàn)閱讀原文

1 限流的簡(jiǎn)介

1.1 限流的作用和場(chǎng)景

對(duì)于后端服務(wù)來(lái)說(shuō)， 他們提供的服務(wù)都有一個(gè)極限的QPS(除代碼邏輯外，也跟機(jī)器配置有關(guān))， 當(dāng)服務(wù)端的壓力超過(guò)這個(gè)極限值的時(shí)候， 服務(wù)端的響應(yīng)性能就會(huì)快速的下降, 然后無(wú)法提供服務(wù)， 所以服務(wù)端需要一個(gè)類(lèi)似于可以限制請(qǐng)求數(shù)的功能， 使服務(wù)端能犧牲掉部分請(qǐng)求， 保證還能處理一定量的請(qǐng)求， 防止服務(wù)端出現(xiàn)壓力瓶頸，無(wú)法處理所有請(qǐng)求。

不過(guò)這個(gè)功能還需要盡量的智能， 在設(shè)計(jì)時(shí)可以根據(jù)流量場(chǎng)景不同來(lái)做有差別的限制， 使其在不影響其它請(qǐng)求的情況下， 實(shí)現(xiàn)部分請(qǐng)求的網(wǎng)絡(luò)流量整形， 達(dá)到減少系統(tǒng)資源消耗的效果， 常見(jiàn)的幾種需要做差別限制的場(chǎng)景如下：

1.2 限流的組件

通過(guò)上述場(chǎng)景可以看到， 在這些場(chǎng)景中限流的作用是差不多的, 一般只涉及到兩個(gè)維度：

• 時(shí)間：對(duì)某個(gè)時(shí)間窗口進(jìn)行限流
• 資源：針對(duì)某個(gè)API或者某個(gè)API的參數(shù)進(jìn)行限流，達(dá)到保護(hù)后方對(duì)應(yīng)的資源。

限流可以保證在某段時(shí)間內(nèi)的某個(gè)資源的請(qǐng)求數(shù)量不會(huì)超過(guò)設(shè)計(jì)值， 達(dá)到保護(hù)系統(tǒng)的作用， 不過(guò)不同場(chǎng)景主要差別是限制的資源維度不一樣， 資源維度的變化從總體服務(wù)到某個(gè)API到某個(gè)API的某個(gè)參數(shù), 資源維度越來(lái)越細(xì)， 而這個(gè)資源維度區(qū)分也就是我們要實(shí)現(xiàn)限流的第一步--流量匹配， 只要流量匹配了， 限流系統(tǒng)就可以開(kāi)始工作了， 一般的限流系統(tǒng)流程圖如下(其中他服務(wù)核心代表微服務(wù)核心)：

限流

流程圖中第一步是規(guī)則匹配， 它會(huì)通過(guò)一個(gè)函數(shù)把流量提取出來(lái)， 當(dāng)做Key， 這個(gè)Key等于某個(gè)資源, 然后判斷這個(gè)Key是否匹配到規(guī)則， 如果命中規(guī)則就開(kāi)始執(zhí)行規(guī)則并結(jié)合這段規(guī)則和限流算法來(lái)判斷該流量是否限流， 如果限流就丟棄或者等待， 如果沒(méi)被限流， 就直接放行。

此外， 流程圖的最下層有一個(gè)很大的Backend， 它可以用來(lái)存儲(chǔ)規(guī)則以及存儲(chǔ)一些限流相關(guān)的計(jì)算變量。其中，限流相關(guān)的計(jì)算變量都是跟時(shí)間相關(guān)的, 且每次都要進(jìn)行讀寫(xiě)， 最好的情況是放在內(nèi)存之中，不過(guò)它不能跟請(qǐng)求綁定在一起， 因?yàn)楦?dāng)前請(qǐng)求的生命周期不一樣， 不能在發(fā)送請(qǐng)求結(jié)束后就把變量回收了， 這些變量也需要有個(gè)容器可以存儲(chǔ), 供不同的請(qǐng)求讀寫(xiě)， 但是在一個(gè)集群服務(wù)中， 每個(gè)機(jī)器都只存儲(chǔ)自己的計(jì)算變量則會(huì)導(dǎo)致多臺(tái)機(jī)器沒(méi)辦法共享數(shù)據(jù)而造成限流失敗。

比如針對(duì)某個(gè)用戶(hù)可以調(diào)用某個(gè)API的規(guī)則是一秒內(nèi)可以請(qǐng)求十次， 目前有十臺(tái)機(jī)器， 他們不會(huì)互相共享自己的限流計(jì)算變量， 那么在最壞的情況下， 用戶(hù)可以在1秒內(nèi)訪問(wèn)100次請(qǐng)求而不被限流， 這樣是達(dá)不了限流的效果的， 所以限流必定是一個(gè)中心化的應(yīng)用。目前兩個(gè)比較主流的限流方案分別是網(wǎng)關(guān)限流和中間件限流， 網(wǎng)關(guān)限流場(chǎng)景下所有入站流量都會(huì)經(jīng)過(guò)網(wǎng)關(guān)這個(gè)單體， 然后由網(wǎng)關(guān)決定是否放行;而中間件限流則是把計(jì)算變量都存在某個(gè)中間件存儲(chǔ)中， 然后每個(gè)服務(wù)的限流組件都可以從中間件實(shí)時(shí)寫(xiě)入和讀取數(shù)據(jù), 其中最常用的中間件是Redis, 因?yàn)镽edis的速度快， 能讓限流組件很快的判斷是否需要限流， 對(duì)機(jī)器的性能開(kāi)銷(xiāo)占比也不是很多， 同時(shí)Redis支持的數(shù)據(jù)結(jié)構(gòu)和功能非常的多， 我們可以很容易的基于它來(lái)實(shí)現(xiàn)不同的限流算法。

至于限流的規(guī)則， 由于它只要寫(xiě)入一次， 后面都是以讀為主， 所以在網(wǎng)關(guān)場(chǎng)景下都存在于內(nèi)存之中， 但在中間件場(chǎng)景下規(guī)則都是存在一個(gè)集中式存儲(chǔ)中， 如Etcd， 然后每個(gè)服務(wù)會(huì)同步集中式存儲(chǔ)的規(guī)則， 并寫(xiě)入到自己的內(nèi)存中。

在實(shí)際的落地要選擇網(wǎng)關(guān)限流還是中間件限流主要還是取決于是服務(wù)的應(yīng)用場(chǎng)景， 比如接口外層都有加一層網(wǎng)關(guān)， 那采用網(wǎng)關(guān)限流即可， 如果是內(nèi)部服務(wù)或者該服務(wù)的通信協(xié)議是自定義的， 則采用中間件方式， 有比較強(qiáng)的自定義性。

在使用Redis下的某些情況下(取決于搭建方式)， 有可能造成數(shù)據(jù)不準(zhǔn)的情況， 但是限流的頻率是允許有些許誤差的， 比如限流的規(guī)則是1秒可以訪問(wèn)100次， 但在某些時(shí)候只實(shí)現(xiàn)了1秒訪問(wèn)110次也是沒(méi)太大關(guān)系的。

1.3 限流算法

上面所說(shuō)的都是一些簡(jiǎn)單的概念， 而限流的核心是在于限流算法的實(shí)現(xiàn)， 常見(jiàn)的限流算法有以下幾種(由于大多數(shù)都限流backend默認(rèn)是Redis， 所以以可以在Redis運(yùn)行的lua代碼示例):

1.3.1 固定窗口

固定窗口的原理比較簡(jiǎn)單，就是將時(shí)間切分成若干個(gè)時(shí)間片，每個(gè)時(shí)間片內(nèi)固定處理若干個(gè)請(qǐng)求。比如限流規(guī)則是10秒內(nèi)最多處理5個(gè)請(qǐng)求， 那么就會(huì)有一個(gè)容器來(lái)統(tǒng)計(jì)這10秒內(nèi)的請(qǐng)求數(shù)， 如果容器的統(tǒng)計(jì)數(shù)量大于5, 那么后續(xù)的請(qǐng)求都會(huì)被拒絕， 然后每隔10秒重置這個(gè)容器的統(tǒng)計(jì)。這種實(shí)現(xiàn)非常簡(jiǎn)單， 但不是非常嚴(yán)謹(jǐn)， 假如限制規(guī)則是1秒限制100個(gè)， 但在最壞的情況下， 在第一個(gè)窗口的0.5秒后到第二個(gè)窗口的0.5秒前的這個(gè)時(shí)間點(diǎn)共計(jì)會(huì)放行200個(gè)請(qǐng)求， 所以固定窗口只適用于一些要求不嚴(yán)格的場(chǎng)景, 通過(guò)下圖的左圖可以看到限流的流程， 通過(guò)下圖的右圖可以看到整個(gè)限流曲線不平滑。

固定窗口

固定窗口的實(shí)現(xiàn)很簡(jiǎn)單， 在Redis中的lua代碼如下:

-- keys為傳入的命令， 其中keys[1]為限流的key  
-- argv為傳入的參數(shù), ARGV[1]為窗口限制量， ARGV[2]為窗口時(shí)間  
local count 
local limit ＝ ARGV[1] 
count = redis.call("incr",KEYS[1]) 
if tonumber(count) == 1 then 
    -- 返回1代表是第一個(gè)， 該key剛被創(chuàng)建， 需要設(shè)置過(guò)期時(shí)間 
    redis.call("expire",KEYS[1], ARGV[2]]) 
end 
return count＜limit  

1.3.2 滑動(dòng)窗口

滑動(dòng)窗口是固定窗口的改進(jìn)方法, 他是通過(guò)增加窗口數(shù)量使限流算法更順滑, 本身從一個(gè)窗口變?yōu)橐粋€(gè)先進(jìn)先出的隊(duì)列， 隊(duì)列的內(nèi)容是更加精細(xì)的窗口，比如原來(lái)是10秒一個(gè)窗口， 現(xiàn)在會(huì)改為1秒一個(gè)窗口， 然后每隔一秒鐘滑動(dòng)一個(gè)窗口。只寫(xiě)入最新的窗口而讀取判斷時(shí)都是取最近10個(gè)窗口， 這樣就可以通過(guò)減小粒度來(lái)讓限流算法更加精細(xì)， 可以看到波動(dòng)幅度會(huì)變小(取決于精細(xì)程度)：

滑動(dòng)窗口的實(shí)現(xiàn)也是很簡(jiǎn)單的， 具體見(jiàn):RPC框架編寫(xiě)實(shí)踐--服務(wù)治理的基石, 在Redis可以采用Zset數(shù)據(jù)結(jié)構(gòu)進(jìn)行實(shí)現(xiàn), 這里就不做代碼示例了。滑動(dòng)窗口是犧牲一定的內(nèi)存來(lái)讓限流變得平滑，窗口數(shù)量越多, 限流速率越精細(xì)， 占用的內(nèi)存就越大， 同時(shí)獲取數(shù)據(jù)時(shí)都是獲取一批窗口的數(shù)據(jù)， 相比于固定窗口來(lái)說(shuō)，它的時(shí)間復(fù)雜度也會(huì)跟著變多(O(k))。

1.3.3 漏桶

漏桶的出現(xiàn)可以完美的解決參差不齊的速率限制問(wèn)題, 漏桶算法的核心原理是進(jìn)入漏桶的請(qǐng)求量不限制， 但能漏出去的速率請(qǐng)求是恒定的， 這樣就能完美的控制請(qǐng)求的速率， 如果桶滿了， 在漏桶里的請(qǐng)求就會(huì)溢出去, 達(dá)到丟棄請(qǐng)求的目的， 如下圖， 整個(gè)請(qǐng)求的速率都是很平滑的， 沒(méi)有多少毛尖：

從圖中可以看到， 漏桶的原理很像一個(gè)FIFO隊(duì)列， 然后有個(gè)定時(shí)器會(huì)以恒定的速率把請(qǐng)求取出來(lái)， 使用Python代碼實(shí)現(xiàn)如下：

import asyncio 
 
# 假設(shè)容量只有10 
import time 
 
leaky_bucket: asyncio.Queue = asyncio.Queue(10) 
loop: asyncio.AbstractEventLoop = asyncio.get_event_loop() 
 
 
async def demo_request(cnt: int) -> None: 
    """模仿請(qǐng)求""" 
    msg: str = f"I'm mock request:{cnt}" 
    future: asyncio.Future = asyncio.Future() 
    try: 
        leaky_bucket.put_nowait(future) 
    except asyncio.QueueFull: 
        # 代表桶滿了， 溢出來(lái)， 該請(qǐng)求要提前拋棄 
        print(f"Fail Request:{msg}") 
    else: 
        # 等待放行 
        await future 
        print(time.time(), msg) 
 
 
def timer() -> None: 
    """定時(shí)放行請(qǐng)求""" 
    try: 
        # 放行該請(qǐng)求 
        future: asyncio.Future = leaky_bucket.get_nowait() 
        future.set_result(True) 
    except asyncio.QueueEmpty: 
        pass 
    # 一秒執(zhí)行一次 
    loop.call_later(1, timer) 
 
timer() 
# 模擬并發(fā)12個(gè)請(qǐng)求 
loop.run_until_complete(asyncio.gather(*[demo_request(i) for i in range(12)])) 

但是， 這樣實(shí)現(xiàn)的漏桶算法依然需要占用一些空間用來(lái)存儲(chǔ)等待放行的請(qǐng)求， 直到放行才被釋放。為了解決空間占用的問(wèn)題， 可以采用GCRA算法， 它從另外一個(gè)角度看起來(lái)跟漏桶算法很像(GRRA應(yīng)該被認(rèn)為是計(jì)量器實(shí)現(xiàn)的漏桶版本， 而不是上面所說(shuō)的隊(duì)列形漏桶), 但很省空間占用， 因?yàn)闊o(wú)論漏桶多大， 它的空間占用都是恒定的, 只需要存漏水速率(可以認(rèn)為是該時(shí)間段可以放行的請(qǐng)求量)以及桶目前的容量即可。

使用GCRA算法之所以能這樣省空間， 主要還是它是基于虛擬調(diào)度實(shí)現(xiàn)的， 它只需要存一個(gè)漏水速率，然后每次有請(qǐng)求進(jìn)來(lái)時(shí)判斷現(xiàn)在可否可以漏水， 如果可以就放行， 如果不可以則判斷桶是否滿， 滿則拋棄請(qǐng)求， 沒(méi)滿則讓請(qǐng)求等待， 直到可以放行為止。常見(jiàn)的GCRA限流實(shí)現(xiàn)一般都考慮使用redis-cell, 它的使用方法如下：

# 第一個(gè)參數(shù)為命令， 第二個(gè)參數(shù)是要限流的key, 第三個(gè)參數(shù)是桶容量， 第四第五綜合起來(lái)為漏桶速率， 第五個(gè)為每次漏多少 
 
# 第一次請(qǐng)求放行， 可以發(fā)現(xiàn)容量變多了一個(gè) 
127.0.0.1:6379> CL.THROTTLE demo_leaky_bucket 2 1 10 1 
1) "0"     # 0表示允許， 1表示拒絕 
2) "3"     # 漏桶容量(會(huì)比輸入的多1) 
3) "2"     # 漏斗剩余空間 
4) "-1"    # 如果拒絕， 需要多長(zhǎng)時(shí)間后重試, 單位秒 
5) "10"    # 多少時(shí)間后，漏桶完全空了, 單位秒 
 
# 第二次請(qǐng)求被放行， 但是漏斗已經(jīng)被占了一個(gè)空位 
127.0.0.1:6379> CL.THROTTLE demo_leaky_bucket 2 1 10 1 
1) "0" 
2) "3" 
3) "1" 
4) "-1" 
5) "18" 
# 第三次請(qǐng)求被放行， 但是漏斗已經(jīng)被占了兩個(gè)個(gè)空位 
127.0.0.1:6379> CL.THROTTLE demo_leaky_bucket 2 1 10 1 
1) "0" 
2) "3" 
3) "0" 
4) "-1" 
5) "27" 
# 第四次請(qǐng)求不被放行， 但是漏斗沒(méi)有空位了 
127.0.0.1:6379> CL.THROTTLE demo_leaky_bucket 2 1 10 1 
1) "1" 
2) "3" 
3) "0" 
4) "6" 
5) "26" 

從命令可以看出, 即使漏斗中還有數(shù)據(jù)沒(méi)漏出去， 返回值得第一個(gè)也還是0， 表示放行， 這樣并不是一個(gè)完善的GCRA。為了實(shí)現(xiàn)一個(gè)完備的GCRA， 我們需要額外的在代碼判斷漏桶是否全空， 如果放行且桶不是全空， 則需要在代碼判斷多久后才能會(huì)為空， 這個(gè)時(shí)間也就是請(qǐng)求的等待放行時(shí)間, 在等待這段時(shí)間后才能放行請(qǐng)求， 如果不是放行， 則直接丟棄請(qǐng)求即可。

不過(guò)， 如果不做任何判斷， 直接使用返回值的第一個(gè)值來(lái)判斷是否放行請(qǐng)求， 那這個(gè)實(shí)現(xiàn)就很像下面所說(shuō)的令牌桶的實(shí)現(xiàn)。

1.3.4 令牌桶

漏桶能很好的控制速率， 使其變得平滑， 但是它沒(méi)辦法應(yīng)對(duì)突發(fā)流量， 比如我們把規(guī)則定義為10秒內(nèi)可以請(qǐng)求10次， 對(duì)于漏桶來(lái)說(shuō)， 它會(huì)控制為1秒放行一個(gè)請(qǐng)求， 如果同時(shí)收到10個(gè)請(qǐng)求時(shí)它則會(huì)分開(kāi)10秒放行每個(gè)請(qǐng)求。然而10秒內(nèi)可以請(qǐng)求10次的含義是10秒內(nèi)總共可以請(qǐng)求10次， 也就是允許在這10秒內(nèi)的某個(gè)瞬間同時(shí)放行10個(gè)請(qǐng)求, 對(duì)于這個(gè)問(wèn)題可以使用令牌桶來(lái)解決, 令牌桶和漏桶很像， 只是漏桶控制的是請(qǐng)求， 令牌桶控制的是令牌發(fā)放速度。

令牌桶算法規(guī)定每個(gè)請(qǐng)求需要從桶里拿到并消耗一個(gè)令牌才可以放行， 拿不到則會(huì)被拋棄， 同時(shí)令牌桶本身會(huì)以恒定的速率產(chǎn)生令牌， 直到桶滿為止， 這樣就可以保證限流的平緩， 同時(shí)又能應(yīng)對(duì)突發(fā)請(qǐng)求， 令牌桶的原理圖和限流曲線圖如下， 其中限流曲線圖表示初始時(shí)桶里面放滿了令牌， 所以放行的請(qǐng)求很多， 隨著令牌被逐漸消耗并消耗光了， 限流的曲率會(huì)穩(wěn)定在一條線上, 也就是令牌的生產(chǎn)速率：

同樣的， 在實(shí)現(xiàn)令牌桶時(shí)為了減少空間的占用， 也會(huì)使用虛擬調(diào)度方法， 只存一個(gè)時(shí)間和容量到內(nèi)存中， 每次收到請(qǐng)求時(shí)都會(huì)根據(jù)請(qǐng)求的時(shí)間和在內(nèi)存中的時(shí)間差值再乘以速率計(jì)算這段時(shí)間應(yīng)該產(chǎn)生的令牌數(shù)量并存到內(nèi)存中， 然后再判斷是否有足夠的令牌來(lái)判斷是否放行請(qǐng)求， 具體的Redis lua代碼實(shí)現(xiàn)如下：

local key = KEYS[1] -- key 
local current_time = redis.call('TIME')[1] -- redis時(shí)間戳 
local interval_per_token = tonumber(ARGV[1]) --每個(gè)單位產(chǎn)生多少個(gè)token 
local max_token = tonumber(ARGV[2]) -- 桶最大的量 
local init_token = tonumber(ARGV[3]) -- 桶初始量 
local tokens 
-- 上次請(qǐng)求時(shí)保留的桶數(shù)據(jù) 
local bucket = redis.call("hmget", key, "last_time", "last_token") 
local last_time= bucket[1] 
local last_token = bucket[2] 
if last_time == false or last_token == false then 
    -- 如果沒(méi)數(shù)據(jù)， 則代表該資源是第一次訪問(wèn)， 進(jìn)行初始化 
    tokens = init_token 
    redis.call('hset', key, 'last_time', current_time) 
else 
    -- 算出間隔時(shí)間 
    local this_interval = current_time - tonumber(last_time) 
    if this_interval > 1 then 
        -- 算出該時(shí)間應(yīng)該產(chǎn)生的令牌 
        local tokens_to_add = math.floor(this_interval * interval_per_token) 
        -- 算出真實(shí)可以擁有的令牌 
        tokens = math.min(last_token + tokens_to_add, max_token) 
        -- 保存數(shù)據(jù) 
        redis.call('hset', key, 'last_time', current_time) 
    else 
        tokens = tonumber(last_token) 
    end 
end 
if tokens < 1 then 
    -- 令牌不夠消費(fèi) 
    redis.call('hset', key, 'last_token', tokens) 
    return -1 
else 
    -- 消費(fèi)令牌并返回令牌數(shù), 代表可以消費(fèi) 
    tokens = tokens - 1 
    redis.call('hset', key, 'last_token', tokens) 
    return tokens 
end 

2.具體實(shí)現(xiàn)

上面說(shuō)完了算法實(shí)現(xiàn)后， 接下來(lái)來(lái)看看該如何結(jié)合算法進(jìn)行實(shí)現(xiàn)， 由于代碼會(huì)隨時(shí)更新，具體源碼更新見(jiàn)：https://github.com/so1n/rap/tree/master/rap/server/plugin/processor/limit

項(xiàng)目的代碼結(jié)構(gòu)如下, 在常見(jiàn)的后端服務(wù)中需要占用空間少， 然后速度盡量快點(diǎn)限流組件， 所以一般只用漏桶或者令牌桶且基于Redis的實(shí)現(xiàn), 這里就不會(huì)去實(shí)現(xiàn)窗口相關(guān)的限流了:

├── backend  # 算法 
│   ├── base.py  # 封裝的協(xié)議 
│   └── redis.py  # 基于redis當(dāng)做banckend的算法實(shí)現(xiàn) 
├── core.py  # 核心判斷代碼， 實(shí)際上是一個(gè)中間流量處理 
├── rule.py  # 規(guī)則聲明 
└── util.py  # 其它小代碼 

首先是rule.py里的規(guī)則類(lèi), 它主要是聲明了限流速率， 初始化token數(shù)量， 最多的tokens數(shù)量以及停用時(shí)間， 其中停用時(shí)間是用來(lái)防止惡意用戶(hù)頻繁刷新， 它的邏輯是當(dāng)漏桶已經(jīng)滿了或者令牌桶沒(méi)有令牌的時(shí)候， 限流組件會(huì)在停用時(shí)間內(nèi)不再提供服務(wù)。

然后就是backend.base.py, 它是一個(gè)限流算法的統(tǒng)一封裝， 代碼如下:

from typing import Any, Coroutine, Union 
 
from rap.server.plugin.processor.limit.rule import Rule 
 
 
class BaseLimitBackend(object): 
    def can_requests(self, key: str, rule: Rule, token_num: int = 1) -> Union[bool, Coroutine[Any, Any, bool]]: 
        raise NotImplementedError 
 
    def expected_time(self, key: str, rule: Rule) -> Union[float, Coroutine[Any, Any, float]]: 
        raise NotImplementedError 

這個(gè)類(lèi)它聲明了兩個(gè)方法， 一個(gè)是can_request, 它會(huì)根據(jù)算法來(lái)判斷是否放行， 如果需要等待， 則會(huì)在這個(gè)方法里進(jìn)行等待， 直到到時(shí)間后才返回放行標(biāo)記, 其中can_request還內(nèi)嵌了一個(gè)block_time的邏輯;另外一個(gè)是expected_time， 用來(lái)獲取下次可用的時(shí)間, 具體的實(shí)現(xiàn)以RedisCellBackend為例子， 它是一個(gè)子類(lèi)。

它的最上層實(shí)現(xiàn)是BaseLimitBackend, 然后就是繼承于BaseLimitBackend的BaseRedisBackend, 這個(gè)組件Redis限流算法的基礎(chǔ)實(shí)現(xiàn)， 主要是實(shí)現(xiàn)了一個(gè)停用時(shí)間的邏輯， 當(dāng)發(fā)現(xiàn)不放行請(qǐng)求的時(shí)候， 會(huì)啟用停用邏輯， 以停用后續(xù)相同key的請(qǐng)求：

class BaseRedisBackend(BaseLimitBackend, ABC): 
    def __init__(self, redis: Union[StrictRedis, StrictRedisCluster]): 
        # 初始化Redis模塊 
        self._redis: "Union[StrictRedis, StrictRedisCluster]" = redis 
 
    async def _block_time_handle(self, key: str, rule: Rule, func: Callable[..., Awaitable[bool]]) -> bool: 
        """處理block_time邏輯""" 
        block_time_key: str = f"{key}:block_time" 
        bucket_block_time: Optional[int] = rule.block_time 
 
        if bucket_block_time is not None and await self._redis.exists(block_time_key): 
            # 啟用block_time邏輯， 且key已經(jīng)存在， 那么直接返回False告訴該請(qǐng)求應(yīng)該被拒絕  
            return False 
 
        # 執(zhí)行正真的判斷是否限流邏輯 
        can_requests: bool = await func() 
 
        if not can_requests and bucket_block_time is not None: 
            # 啟用block_time邏輯且被限流時(shí), 正式啟用block time邏輯  
            await self._redis.set(block_time_key, bucket_block_time, ex=bucket_block_time) 
 
        return can_requests 

接著就是繼承于BaseRedisBackend的BaseRedisCellBackend, 它主要是提供一個(gè)命令調(diào)用的封裝以及獲取還有多久后才能請(qǐng)求的封裝：

class BaseRedisCellBackend(BaseRedisBackend): 
    """ 
    use redis-cell module 
    learn more:https://github.com/brandur/redis-cell 
 
    input: CL.THROTTLE user123 15 30 60 1 
        # param  |  desc 
        # user123 key 
        # 15 maxburst 
        # 30 token 
        # 60 seconds 
        # 1 apply 1token 
    output: 
        1) (integer) 0        # is allowed 
        2) (integer) 16       # total bucket num 
        3) (integer) 15       # the remaining limit of the key. 
        4) (integer) -1       # the number of seconds until the user should retry, 
                              #   and always -1 if the action was allowed. 
        5) (integer) 2        # The number of seconds until the limit will reset to its maximum capacity 
    """ 
 
    async def _call_cell(self, key: str, rule: Rule, token_num: int = 1) -> List[int]: 
        """調(diào)用redis_cell""" 
        result: List[int] = await self._redis.execute_command( 
            "CL.THROTTLE", key, rule.max_token - 1, rule.gen_token, int(rule.total_second), token_num 
        ) 
        return result 
 
    def expected_time(self, key: str, rule: Rule) -> Union[float, Coroutine[Any, Any, float]]: 
        """獲取下次可請(qǐng)求時(shí)間""" 
        async def _expected_time() -> float: 
            block_time_key: str = key + ":block_time" 
            block_time = await self._redis.get(block_time_key) 
            if block_time: 
                return await self._redis.ttl(block_time_key) 
 
            result: List[int] = await self._call_cell(key, rule, 0) 
            return float(max(result[3], 0)) 
 
        return _expected_time() 

最后就是真正的對(duì)外使用的限流組件實(shí)現(xiàn)， 這個(gè)實(shí)現(xiàn)是基于漏桶算法的， 它繼承于BaseRedisCellBackend(另外一個(gè)繼承于BaseRedisCellBackend的實(shí)現(xiàn)是基于令牌桶算法的， 可以通過(guò)源碼了解), 可以看到非常的簡(jiǎn)單， 本質(zhì)上是基于redis-cell的返回判斷是否放行。

class RedisCellBackend(BaseRedisCellBackend): 
 
    def can_requests(self, key: str, rule: Rule, token_num: int = 1) -> Union[bool, Coroutine[Any, Any, bool]]: 
        """通過(guò)redis-cell判斷是否可以請(qǐng)求，以及是否需要休眠等待， 如果需要?jiǎng)t休眠固定的時(shí)間后再放行""" 
        async def _can_requests() -> bool: 
            result: List[int] = await self._call_cell(key, rule, token_num) 
            can_requests: bool = result[0] == 0 
            if can_requests and result[4]: 
                await asyncio.sleep(result[4]) 
            return can_requests 
 
        return self._block_time_handle(key, rule, _can_requests) 

了解完算法的實(shí)現(xiàn)后， 接下來(lái)就是核心的判斷邏輯, 具體見(jiàn)注釋:

class LimitProcessor(BaseProcessor): 
    def __init__(self, backend: BaseLimitBackend, rule_list: List[Tuple[RULE_FUNC_TYPE, Rule]]): 
        """初始化規(guī)則和算法邏輯， 這里的規(guī)則之所以是使用傳參的方式是僅供參考， 后續(xù)整個(gè)框架的配置都會(huì)抽離成一個(gè)config， 供其它組件調(diào)用""" 
        self._backend: BaseLimitBackend = backend 
        self._rule_list: List[Tuple[RULE_FUNC_TYPE, Rule]] = rule_list 
 
    async def process_request(self, request: Request) -> Request: 
        # not limit client event 
        if request.msg_type == constant.CLIENT_EVENT: 
            # 屏蔽event請(qǐng)求 
            return request 
 
        for func, rule in self._rule_list: 
            # 獲取該請(qǐng)求的key 
            if inspect.iscoroutinefunction(func): 
                key, is_ignore_limit = await func(request)  # type: ignore 
            else: 
                key, is_ignore_limit = func(request) 
            if is_ignore_limit: 
                # 如果該請(qǐng)求不應(yīng)該限流， 直接跳過(guò)限流邏輯 
                return request 
            if key: 
                # 匹配到key， 進(jìn)入限流邏輯 
                break 
        else: 
            raise TooManyRequest() 
 
        # 通過(guò)backend判斷是否限流 
        key = f"rap:processor:{self.__class__.__name__}:{key}" 
        can_requests: Union[bool, Awaitable[bool]] = self._backend.can_requests(key, rule) 
        if inspect.isawaitable(can_requests): 
            can_requests = await can_requests  # type: ignore 
        if not can_requests: 
            # 如果被限流， 返回異常， 并告知要何時(shí)后才可以再次請(qǐng)求 
            expected_time: Union[float, Awaitable[float]] = self._backend.expected_time(key, rule) 
            if inspect.isawaitable(expected_time): 
                expected_time = await expected_time  # type: ignore 
            raise TooManyRequest(extra_msg=f"expected time: {expected_time}") 
        return request 

至此， 整個(gè)限流邏輯實(shí)現(xiàn)完畢, 本章內(nèi)容完。

3.其它碎碎念

3.1.熱點(diǎn)參數(shù)實(shí)現(xiàn)

由于大部分的限流實(shí)現(xiàn)的backend都只要依賴(lài)于Redis, 所以代碼倉(cāng)里面只有Redis一種類(lèi)型的backend，但是有一些限流實(shí)現(xiàn)需要依賴(lài)于一些特殊的backend，比如熱點(diǎn)參數(shù)限流, 還有蜜罐之類(lèi)的場(chǎng)景。

以熱點(diǎn)參數(shù)限流場(chǎng)景為例子， 熱點(diǎn)參數(shù)是一個(gè)寫(xiě)大于讀的應(yīng)用場(chǎng)景， 而且跟時(shí)間強(qiáng)相關(guān)， 所以選用時(shí)序數(shù)據(jù)庫(kù)做backend，之前選用過(guò)Graphite當(dāng)做backend, 具體實(shí)現(xiàn)如圖后端服務(wù)會(huì)把每次請(qǐng)求參數(shù)都記錄到時(shí)序數(shù)據(jù)庫(kù)中， 并使用一個(gè)定時(shí)腳本每隔一段時(shí)間把最近的熱點(diǎn)參數(shù)數(shù)據(jù)拉取到緩存中， 供后端服務(wù)的限流組件判斷是否該放行。其中， 這個(gè)間隔一般控制在1秒左右， 所以這是一個(gè)近實(shí)時(shí)的實(shí)現(xiàn), 具體的實(shí)現(xiàn)圖如下：

當(dāng)請(qǐng)求進(jìn)來(lái)的時(shí)候， 限流中間件會(huì)通過(guò)異步的方法把數(shù)據(jù)記錄到時(shí)序數(shù)據(jù)庫(kù)中， 比如一個(gè)請(qǐng)求為http://127.0.0.1:80?q=1&b=2，中間件就會(huì)發(fā)送一個(gè)以{prefix}.hot_param.b=2&b=2為key, value為1的標(biāo)準(zhǔn)Statsd的count類(lèi)型數(shù)據(jù)到Statsd組件中。

這個(gè)Key采用標(biāo)準(zhǔn)的Statsd命令, 以.分割有三個(gè)值， 第一個(gè)是前綴它與業(yè)務(wù)相關(guān), 如業(yè)務(wù)名， 函數(shù)名，namespace等等; 第二個(gè)是代表是熱點(diǎn)參數(shù)的業(yè)務(wù);第三個(gè)是參數(shù)Key, 這里以&為分割號(hào)， 然后按Key順序排序，重新拼接為一個(gè)字符串， 這樣即使請(qǐng)求時(shí)順序不一致也能識(shí)別到時(shí)同種請(qǐng)求。

Statsd組件收到了數(shù)據(jù)后會(huì)自行進(jìn)行統(tǒng)計(jì)， 統(tǒng)計(jì)一個(gè)時(shí)間區(qū)間都數(shù)據(jù)并寫(xiě)入到Graphite中， 然后通過(guò)定時(shí)腳本使用Graphite API拉取統(tǒng)計(jì)次數(shù)大于條件的數(shù)據(jù)寫(xiě)入到Redis緩存中， 其中Statsd組件的時(shí)間區(qū)間和定時(shí)腳本的定時(shí)時(shí)間都會(huì)控制在一秒左右， 所以這是一個(gè)近實(shí)時(shí)的實(shí)現(xiàn)， 在計(jì)算性能消耗和實(shí)現(xiàn)效果直接做取舍。

數(shù)據(jù)寫(xiě)入到Redis后， 這個(gè)寫(xiě)入數(shù)據(jù)和統(tǒng)計(jì)的異步流程就結(jié)束了， 中間件在記錄數(shù)據(jù)后， 會(huì)通過(guò)Redis判斷是否是熱點(diǎn)參數(shù)， 并根據(jù)規(guī)則判斷是否放行， 到了這里就跟上面的限流流程差不多了。

3.2.限流算法拓展

限流算法不止是用于算法， 也可以用于別的地方，比如有一些游戲活動(dòng)，體力值滿時(shí)為5, 然后玩家每次出發(fā)活動(dòng)會(huì)減少1個(gè)體力值, 然后可以使用限流算法每隔一個(gè)固定時(shí)間則增加一個(gè)體力值等等。在遇到業(yè)務(wù)需求有跟時(shí)間相關(guān)且像上述所說(shuō)的體力值會(huì)恢復(fù)的情況時(shí)， 可以往限流算法思考。