Django REST framework 限流功能的使用
先說一個限流這個概念，最早接觸這個概念是在前端。真實的業務場景是在搜索框中輸入文字進行搜索時，并不希望每輸一個字符都去調用后端接口，而是有停頓后才真正的調用接口。這個功能很有必要，一方面減少前端請求與渲染的壓力，同時減輕后端接口訪問的壓力。類似前端的功能的代碼如下：
// 前端函數限流示例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 的限流。
1. DRF 中的限流項目配置
# demo/settings.pyREST_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 秒后。
2. 限流進階配置上述演示的限流配置適用于對用戶的限流，比如我換個用戶繼續訪問，依然是有兩次的機會。
$ 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如果你愿意'}
分別介紹一下三種限流類
AnonRateThrottle 適用于任何用戶對接口訪問的限制 UserRateThrottle 適用于請求認證結束后對接口訪問的限制 ScopedRateThrottle 適用于對多個接口訪問的限制所以三種不同的類適用于不同的業務場景，具體使用根據不同的業務場景選擇，通過配置相對應 scope 的頻率的配置就可以達到預期的效果。
3. 限流思路分析試想一下如果是你編碼實現這個需求應該怎么實現？
其實這個功能不難，核心的參數就是 時間、次數、使用范圍，下面演示對函數調用次數的限制。
from functools import wrapsTOTAL_RATE = 2FUNC_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 resultelse: # 不在計數限制的函數不受限制 return func() return wrapper@rate_countdef test1(): pass@rate_countdef test2(): print('test2') passif __name__ == '__main__': try:test2()test2()test1()test1()test1() except Exception as e:print(e) test2() test2() '''test2test2 函數 test1 調用次數為： 1 函數 test1 調用次數為： 2test1函數調用超過設定次數test2test2'''
這里實現了對函數調用次數的監控同時設置了能夠使用該功能的函數。當函數調用次數超過設定閥值久拋出異常。只是這里沒有對時間做限制。
4. 源碼分析剛才分析了如何實現對函數調用次數的限制，對于一個請求來說可能會復雜一點，下面就看看 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 Trueself.key = self.get_cache_key(request, view)if self.key is None: return Trueself.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.pkelse: # 考慮到用戶沒有認證的情況 與 AnonRateThrottle 中 key 一致 ident = self.get_ident(request)# 根據設置的范圍構建緩存的 keyreturn self.cache_format % { ’scope’: self.scope, ’ident’: ident}
綜上所述：
核心的判斷邏輯依舊是緩存中獲取每個用戶調用次數，根據范圍與時間判斷是否超過設置定的閥值。 不同類型的限流，在緩存 key 的設計上會有區別，默認的 key 為請求中REMOTE_ADDR。 5. 其它注意事項 因為這里的實現用到緩存，所以需要注意在多實例部署的情況下需要配置統一的緩存服務（默認的緩存為 Django 基于內存實現的）。 緩存服務的重啟可能會導致已有的計數清零，如果有較強的業務邏輯需要，還請自己實現限流的邏輯。 如果是自定義的用戶表，需要重寫緩存中 get_cache_key 的邏輯。 如果需要統計分析用戶被限流情況也是需要重新設計限流的邏輯。 限流的邏輯在生產環境中慎用，因為會限制用戶使用產品，對用戶不夠友好。參考資料DRF 限流Django 緩存
以上就是Django REST framework 限流功能的使用的詳細內容，更多關于Django REST framework 限流功能的資料請關注好吧啦網其它相關文章！
相關文章：
網公網安備