先說一個限流這個概念，最早接觸這個概念是在前端。真實的業務場景是在搜索框中輸入文字進行搜索時，并不希望每輸一個字符都去調用后端接口，而是有停頓后才真正的調用接口。這個功能很有必要，一方面減少前端請求與渲染的壓力，同時減輕后端接口訪問的壓力。類似前端的功能的代碼如下：
// 前端函數限流示例
function throttle(fn, delay) {
var timer;
return function () {
var _this = this;
var args = arguments;
if (timer) {
return;
}
timer = setTimeout(function () {
fn.apply(_this, args);
timer = null;
}, delay)
}
}
但是后端的限流從目的上來說與前端類似，但是實現上會有所不同，讓我們看看 DRF 的限流。
項目配置
# demo/settings.py
REST_FRAMEWORK = {
# ...
'DEFAULT_THROTTLE_CLASSES': (
'rest_framework.throttling.AnonRateThrottle',
'rest_framework.throttling.UserRateThrottle',
'rest_framework.throttling.ScopedRateThrottle',
),
'DEFAULT_THROTTLE_RATES': {
'anon': '10/day',
'user': '2/day'
},
}
# article/views.py
# 基于ViewSet的限流
class ArticleViewSet(viewsets.ModelViewSet, ExceptionMixin):
"""
允許用戶查看或編輯的API路徑。
"""
queryset = Article.objects.all()
# 使用默認的用戶限流
throttle_classes = (UserRateThrottle,)
serializer_class = ArticleSerializer
# 基于view的限流
@throttle_classes([UserRateThrottle])
因為我配置的用戶每天只能請求兩次，所以在請求第三次之后就會給出 429 Too Many Requests的異常，具體的異常信息為下一次可用時間為 86398 秒后。
上述演示的限流配置適用于對用戶的限流，比如我換個用戶繼續訪問，依然是有兩次的機會。
$ curl -H 'Accept: application/json; indent=4' -u root:root http://127.0.0.1:8000/api/article/1/
{
"id": 1,
"creator": "admin",
"tag": "現代詩",
"title": "如果",
"content": "今生今世 永不再將你想起\n除了\n除了在有些個\n因落淚而濕潤的夜里 如果\n如果你愿意"
}
分別介紹一下三種限流類
所以三種不同的類適用于不同的業務場景，具體使用根據不同的業務場景選擇，通過配置相對應 scope 的頻率的配置就可以達到預期的效果。
試想一下如果是你編碼實現這個需求應該怎么實現？
其實這個功能不難，核心的參數就是 時間、次數、使用范圍，下面演示對函數調用次數的限制。
from functools import wraps
TOTAL_RATE = 2
FUNC_SCOPE = ['test', 'test1']
def rate_count(func):
func_num = {
# 需要注意函數名不能重復
func.__name__: 0
}
@wraps(func)
def wrapper():
if func.__name__ in FUNC_SCOPE:
if func_num[func.__name__] >= TOTAL_RATE:
raise Exception(f"{func.__name__}函數調用超過設定次數")
result = func()
func_num[func.__name__] += 1
print(f" 函數 {func.__name__} 調用次數為： {func_num[func.__name__]}")
return result
else:
# 不在計數限制的函數不受限制
return func()
return wrapper
@rate_count
def test1():
pass
@rate_count
def test2():
print("test2")
pass
if __name__ == "__main__":
try:
test2()
test2()
test1()
test1()
test1()
except Exception as e:
print(e)
test2()
test2()
"""
test2
test2
函數 test1 調用次數為： 1
函數 test1 調用次數為： 2
test1函數調用超過設定次數
test2
test2
"""
這里實現了對函數調用次數的監控同時設置了能夠使用該功能的函數。當函數調用次數超過設定閥值久拋出異常。只是這里沒有對時間做限制。
剛才分析了如何實現對函數調用次數的限制，對于一個請求來說可能會復雜一點，下面就看看 DRF 如何實現的：
class SimpleRateThrottle(BaseThrottle):
# ......
def allow_request(self, request, view):
"""
Implement the check to see if the request should be throttled.
On success calls `throttle_success`.
On failure calls `throttle_failure`.
"""
if self.rate is None:
return True
self.key = self.get_cache_key(request, view)
if self.key is None:
return True
self.history = self.cache.get(self.key, [])
self.now = self.timer()
# 根據設置時間的限制改變請求次數的緩存
while self.history and self.history[-1] = self.now - self.duration:
self.history.pop()
# 核心邏輯就是這里判斷請求次數
if len(self.history) >= self.num_requests:
return self.throttle_failure()
return self.throttle_success()
# ......
class UserRateThrottle(SimpleRateThrottle):
"""
Limits the rate of API calls that may be made by a given user.
The user id will be used as a unique cache key if the user is
authenticated. For anonymous requests, the IP address of the request will
be used.
"""
scope = 'user'
def get_cache_key(self, request, view):
if request.user.is_authenticated:
ident = request.user.pk
else:
# 考慮到用戶沒有認證的情況 與 AnonRateThrottle 中 key 一致
ident = self.get_ident(request)
# 根據設置的范圍構建緩存的 key
return self.cache_format % {
'scope': self.scope,
'ident': ident
}
綜上所述：
DRF 限流
Django 緩存
以上就是Django REST framework 限流功能的使用的詳細內容，更多關于Django REST framework 限流功能的資料請關注腳本之家其它相關文章！
