一日一技:FastAPI 介面限流

語言: CN / TW / HK

今天分享介面限流。

如果沒有介面限流,可能會導致伺服器負載不平衡,暴力破解密碼,惡意請請求,導致伺服器額外費用,拒絕服務攻擊等。

因此做好介面限流很有必要。

怎麼做介面限流呢?常見的介面限流演算法有 4 種:

1、固定視窗計數器

比如說每小時限制請求 10 次,超過 10 次的直接丟棄。它有個缺點,就是有時會超過 10 次,最多達到 2 倍。比如說固定視窗為整點,8 點 50 到 9 點之間傳送了 10 個請求,9 點 到 9 點 10 分又傳送了 10 個請求,雖說都被放行,但 8 點 50 到 9 點 50 這一個小時內,傳送了 20 個請求。

2、滑動視窗計數器

這個解決了 1 的問題,但是時間區間的精度劃分越高,演算法所需的空間容量就越大。

3、漏桶演算法

漏桶演算法多使用佇列實現,服務的請求會存到佇列中,服務的提供方則按照固定的速率從佇列中取出請求並執行,過多的請求則放在佇列中排隊或直接拒絕。漏桶演算法的缺陷也很明顯,當短時間內有大量的突發請求時,即便此時伺服器沒有任何負載,每個請求也都得在佇列中等待一段時間才能被響應。

4、令牌桶演算法

令牌以固定速率生成。生成的令牌放入令牌桶中存放,如果令牌桶滿了則多餘的令牌會直接丟棄,當請求到達時,會嘗試從令牌桶中取令牌,取到了令牌的請求可以執行。如果桶空了,那麼嘗試取令牌的請求會被直接丟棄。令牌桶演算法既能夠將所有的請求平均分佈到時間區間內,又能接受伺服器能夠承受範圍內的突發請求。

可能有朋友會問,為啥不根據 IP 地址進行限流?其實做是可以做的,只是不那麼主流。

既然要根據 IP 地址進行限流,那就會產生兩個問題,一是 IP 地址的儲存就是一個問題,如果介面是叢集,你還要將 IP 地址儲存在一個集中的資料庫裡,最好是 redis。二是會誤傷正常請求,因為一個大的區域網,其出口 IP 是一個,那麼限制了這個 IP 的請求,可能導致正常使用者被困。

以上 4 種方法中,最簡單實用的就是滑動視窗計數器。

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 個方法:

1、 slowapi [2]

slowapi 是有人根據 flask-limiter 改寫的,計數器預設儲存在記憶體中,具體用法如下:

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]

需要一個 redis 來儲存計數器:

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]

比如超過每秒五次訪問限制後,阻止特定使用者 60 秒:

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

以上推薦 slowapi,無它,因為星是目前最多的。

參考資料

[1]自帶限流: https://www.django-rest-framework.org/api-guide/throttling/

[2]slowapi: https://github.com/laurents/slowapi

[3]fastapi-limiter: https://github.com/long2ice/fastapi-limiter

[4]asgi-ratelimit: https://github.com/abersheeran/asgi-ratelimit

未聞 Code·知識星球開放啦!

一對一答疑爬蟲相關問題

職業生涯諮詢

面試經驗分享

每週直播分享

......

未聞 Code·知識星球期待與你相見~

一二線大廠在職員工

十多年碼齡的程式設計老鳥

國內外高校在讀學生

中小學剛剛入門的新人

“未聞 Code技術交流群” 等你來!

入群方式:新增微信“mekingname”,備註“粉絲群”(謝絕廣告黨,非誠勿擾!)