一日一技：FastAPI 接口限流
今天分享接口限流。
如果沒有接口限流，可能會導致服務器負載不平衡，暴力破解密碼，惡意請請求，導致服務器額外費用，拒絕服務攻擊等。
因此做好接口限流很有必要。
怎麼做接口限流呢？常見的接口限流算法有 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
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/nXD_sJbmhEUvgh5bAJuRgA