Django REST framework 限流功能的使用
正文開始
先說一個限流這個概念,最早接觸這個概念是在前端。真實的業(yè)務場景是在搜索框中輸入文字進行搜索時,并不希望每輸一個字符都去調(diào)用后端接口,而是有停頓后才真正的調(diào)用接口。這個功能很有必要,一方面減少前端請求與渲染的壓力,同時減輕后端接口訪問的壓力。類似前端的功能的代碼如下:
// 前端函數(shù)限流示例 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) } }
但是后端的限流從目的上來說與前端類似,但是實現(xiàn)上會有所不同,讓我們看看 DRF 的限流。
1. 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 秒后。
2. 限流進階配置
上述演示的限流配置適用于對用戶的限流,比如我換個用戶繼續(xù)訪問,依然是有兩次的機會。
$ curl -H 'Accept: application/json; indent=4' -u root:root http://127.0.0.1:8000/api/article/1/ { "id": 1, "creator": "admin", "tag": "現(xiàn)代詩", "title": "如果", "content": "今生今世 永不再將你想起\n除了\n除了在有些個\n因落淚而濕潤的夜里 如果\n如果你愿意" }
分別介紹一下三種限流類
- AnonRateThrottle 適用于任何用戶對接口訪問的限制
- UserRateThrottle 適用于請求認證結(jié)束后對接口訪問的限制
- ScopedRateThrottle 適用于對多個接口訪問的限制
所以三種不同的類適用于不同的業(yè)務場景,具體使用根據(jù)不同的業(yè)務場景選擇,通過配置相對應 scope 的頻率的配置就可以達到預期的效果。
3. 限流思路分析
試想一下如果是你編碼實現(xiàn)這個需求應該怎么實現(xiàn)?
其實這個功能不難,核心的參數(shù)就是 時間、次數(shù)、使用范圍,下面演示對函數(shù)調(diào)用次數(shù)的限制。
from functools import wraps TOTAL_RATE = 2 FUNC_SCOPE = ['test', 'test1'] def rate_count(func): func_num = { # 需要注意函數(shù)名不能重復 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__}函數(shù)調(diào)用超過設定次數(shù)") result = func() func_num[func.__name__] += 1 print(f" 函數(shù) {func.__name__} 調(diào)用次數(shù)為: {func_num[func.__name__]}") return result else: # 不在計數(shù)限制的函數(shù)不受限制 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 函數(shù) test1 調(diào)用次數(shù)為: 1 函數(shù) test1 調(diào)用次數(shù)為: 2 test1函數(shù)調(diào)用超過設定次數(shù) test2 test2 """
這里實現(xiàn)了對函數(shù)調(diào)用次數(shù)的監(jiān)控同時設置了能夠使用該功能的函數(shù)。當函數(shù)調(diào)用次數(shù)超過設定閥值久拋出異常。只是這里沒有對時間做限制。
4. 源碼分析
剛才分析了如何實現(xiàn)對函數(shù)調(diào)用次數(shù)的限制,對于一個請求來說可能會復雜一點,下面就看看 DRF 如何實現(xiàn)的:
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() # 根據(jù)設置時間的限制改變請求次數(shù)的緩存 while self.history and self.history[-1] <= self.now - self.duration: self.history.pop() # 核心邏輯就是這里判斷請求次數(shù) 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) # 根據(jù)設置的范圍構(gòu)建緩存的 key return self.cache_format % { 'scope': self.scope, 'ident': ident }
綜上所述:
- 核心的判斷邏輯依舊是緩存中獲取每個用戶調(diào)用次數(shù),根據(jù)范圍與時間判斷是否超過設置定的閥值。
- 不同類型的限流,在緩存 key 的設計上會有區(qū)別,默認的 key 為請求中REMOTE_ADDR。
5. 其它注意事項
- 因為這里的實現(xiàn)用到緩存,所以需要注意在多實例部署的情況下需要配置統(tǒng)一的緩存服務(默認的緩存為 Django 基于內(nèi)存實現(xiàn)的)。
- 緩存服務的重啟可能會導致已有的計數(shù)清零,如果有較強的業(yè)務邏輯需要,還請自己實現(xiàn)限流的邏輯。
- 如果是自定義的用戶表,需要重寫緩存中 get_cache_key 的邏輯。
- 如果需要統(tǒng)計分析用戶被限流情況也是需要重新設計限流的邏輯。
- 限流的邏輯在生產(chǎn)環(huán)境中慎用,因為會限制用戶使用產(chǎn)品,對用戶不夠友好。
參考資料
以上就是Django REST framework 限流功能的使用的詳細內(nèi)容,更多關(guān)于Django REST framework 限流功能的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
python append、extend與insert的區(qū)別
這篇文章主要介紹了python append、extend與insert的區(qū)別的相關(guān)資料,初學者對這幾個概念經(jīng)常搞混,這里就幫大家理清楚,需要的朋友可以參考下2016-10-10使用Pyhton集合set()實現(xiàn)成果查漏的例子
今天小編就為大家分享一篇使用Pyhton集合set()實現(xiàn)成果查漏的例子,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11解決pip安裝tensorflow中出現(xiàn)的no module named tensorflow.python 問題方法
這篇文章主要介紹了解決pip安裝tensorflow中出現(xiàn)的no module named tensorflow.python 問題方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2021-02-02TensorFlow2基本操作之 張量排序 填充與復制 查找與替換
這篇文章主要介紹了TensorFlow2基本操作之 張量排序 填充與復制 查找與替換,本文給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-09-09