FastAPI 接口限流

你好，我是征哥，今天分享接口限流。

如果沒(méi)有接口限流，可能會(huì)導(dǎo)致服務(wù)器負(fù)載不平衡，暴力破解密碼，惡意請(qǐng)請(qǐng)求，導(dǎo)致服務(wù)器額外費(fèi)用，拒絕服務(wù)攻擊等。

因此做好接口限流很有必要。

怎么做接口限流呢？常見(jiàn)的接口限流算法有 4 種：

1、固定窗口計(jì)數(shù)器

比如說(shuō)每小時(shí)限制請(qǐng)求 10 次，超過(guò) 10 次的直接丟棄。它有個(gè)缺點(diǎn)，就是有時(shí)會(huì)超過(guò) 10 次，最多達(dá)到 2 倍。比如說(shuō)固定窗口為整點(diǎn)，8 點(diǎn) 50 到 9 點(diǎn)之間發(fā)送了 10 個(gè)請(qǐng)求，9 點(diǎn) 到 9 點(diǎn) 10 分又發(fā)送了 10 個(gè)請(qǐng)求，雖說(shuō)都被放行，但 8 點(diǎn) 50 到 9 點(diǎn) 50 這一個(gè)小時(shí)內(nèi)，發(fā)送了 20 個(gè)請(qǐng)求。

2、滑動(dòng)窗口計(jì)數(shù)器

這個(gè)解決了 1 的問(wèn)題，但是時(shí)間區(qū)間的精度劃分越高，算法所需的空間容量就越大。

3、漏桶算法

漏桶算法多使用隊(duì)列實(shí)現(xiàn)，服務(wù)的請(qǐng)求會(huì)存到隊(duì)列中，服務(wù)的提供方則按照固定的速率從隊(duì)列中取出請(qǐng)求并執(zhí)行，過(guò)多的請(qǐng)求則放在隊(duì)列中排隊(duì)或直接拒絕。漏桶算法的缺陷也很明顯，當(dāng)短時(shí)間內(nèi)有大量的突發(fā)請(qǐng)求時(shí)，即便此時(shí)服務(wù)器沒(méi)有任何負(fù)載，每個(gè)請(qǐng)求也都得在隊(duì)列中等待一段時(shí)間才能被響應(yīng)。

4、令牌桶算法

令牌以固定速率生成。生成的令牌放入令牌桶中存放，如果令牌桶滿了則多余的令牌會(huì)直接丟棄，當(dāng)請(qǐng)求到達(dá)時(shí)，會(huì)嘗試從令牌桶中取令牌，取到了令牌的請(qǐng)求可以執(zhí)行。如果桶空了，那么嘗試取令牌的請(qǐng)求會(huì)被直接丟棄。令牌桶算法既能夠?qū)⑺械恼?qǐng)求平均分布到時(shí)間區(qū)間內(nèi)，又能接受服務(wù)器能夠承受范圍內(nèi)的突發(fā)請(qǐng)求。

可能有朋友會(huì)問(wèn)，為啥不根據(jù) IP 地址進(jìn)行限流？其實(shí)做是可以做的，只是不那么主流。

既然要根據(jù) IP 地址進(jìn)行限流，那就會(huì)產(chǎn)生兩個(gè)問(wèn)題，一是 IP 地址的保存就是一個(gè)問(wèn)題，如果接口是集群，你還要將 IP 地址保存在一個(gè)集中的數(shù)據(jù)庫(kù)里，最好是 redis。二是會(huì)誤傷正常請(qǐng)求，因?yàn)橐粋€(gè)大的局域網(wǎng)，其出口 IP 是一個(gè)，那么限制了這個(gè) IP 的請(qǐng)求，可能導(dǎo)致正常用戶被困。

以上 4 種方法中，最簡(jiǎn)單實(shí)用的就是滑動(dòng)窗口計(jì)數(shù)器。

Django REST Framework 自帶限流^[1]:

REST_FRAMEWORK?=?{
????'DEFAULT_THROTTLE_CLASSES':?[
????????'rest_framework.throttling.AnonRateThrottle',
????????'rest_framework.throttling.UserRateThrottle'
????],
????'DEFAULT_THROTTLE_RATES':?{
????????'anon':?'100/day',
????????'user':?'1000/day'
????}
}

這里分享一下 FastAPI 限流的 3 個(gè)方法：

1、slowapi^[2]

slowapi 是有人根據(jù) flask-limiter 改寫(xiě)的，計(jì)數(shù)器默認(rèn)保存在內(nèi)存中，具體用法如下：

from?fastapi?import?FastAPI
from?slowapi.errors?import?RateLimitExceeded
from?slowapi?import?Limiter,?_rate_limit_exceeded_handler
from?slowapi.util?import?get_remote_address


limiter?=?Limiter(key_func=get_remote_address)
app?=?FastAPI()
app.state.limiter?=?limiter
app.add_exception_handler(RateLimitExceeded,?_rate_limit_exceeded_handler)

@app.get("/home")
@limiter.limit("5/minute")
async?def?homepage(request:?Request):
????return?PlainTextResponse("test")

@app.get("/mars")
@limiter.limit("5/minute")
async?def?homepage(request:?Request,?response:?Response):
????return?{"key":?"value"}

2、 fastapi-limiter^[3]

需要一個(gè) redis 來(lái)保存計(jì)數(shù)器：

import?aioredis
import?uvicorn
from?fastapi?import?Depends,?FastAPI

from?fastapi_limiter?import?FastAPILimiter
from?fastapi_limiter.depends?import?RateLimiter

app?=?FastAPI()


@app.on_event("startup")
async?def?startup():
????redis?=?await?aioredis.create_redis_pool("redis://localhost")
????FastAPILimiter.init(redis)


@app.get("/",?dependencies=[Depends(RateLimiter(times=2,?seconds=5))])
async?def?index():
????return?{"msg":?"Hello?World"}


if?__name__?==?"__main__":
????uvicorn.run("main:app",?debug=True,?reload=True)

3、asgi-ratelimit^[4]

比如超過(guò)每秒五次訪問(wèn)限制后，阻止特定用戶 60 秒:

app.add_middleware(
????RateLimitMiddleware,
????authenticate=AUTH_FUNCTION,
????backend=RedisBackend(),
????config={
????????r"^/user":?[Rule(second=5,?block_time=60)],
????},
)

以上推薦 slowapi，無(wú)它，因?yàn)樾鞘悄壳白疃嗟摹?/p>

最后的話

本文分享了常見(jiàn)的接口限流技術(shù)，希望有幫助，還請(qǐng)點(diǎn)在看、轉(zhuǎn)發(fā)，幫助更多的朋友。

參考資料