正文开始
先说一个限流这个概念，最早接触这个概念是在前端。真实的业务场景是在搜索框中输入文字进行搜索时，并不希望每输一个字符都去调用后端接口，而是有停顿后才真正的调用接口。这个功能很有必要，一方面减少前端请求与渲染的压力，同时减轻后端接口访问的压力。类似前端的功能的代码如下：
// 前端函数限流示例
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.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. 限流进阶配置
上述演示的限流配置适用于对用户的限流，比如我换个用户继续访问，依然是有两次的机会。
$ 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 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__] <p style="color:transparent">本文来源gao!%daima.com搞$代*!码$网3</p>+= 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
"""
搞代码网（gaodaima.com）提供的所有资源部分来自互联网，如果有侵犯您的版权或其他权益，请说明详细缘由并提供版权或权益证明然后发送到邮箱[email protected]，我们会在看到邮件的第一时间内为您处理，或直接联系QQ：872152909。本网站采用BY-NC-SA协议进行授权
转载请注明原文链接：Django REST framework 限流功能的使用
转载请注明原文链接：Django REST framework 限流功能的使用
