一、自定义限流
限流组件又叫做频率组件，用于控制客户端可以对API进行的请求频率，比如说1分钟访问3次，如果在1分钟内超过3次就对客户端进行限制。
1、自定义限流
假设现在对一个API访问，在30s内访问不能超过3次，应该如何实现？
VISIT_RECORD = {} #定义全局变量，用于存放访问记录
class VisitThrottle(object):
def __init__(self):
#用于await计算剩余访问时间
self.history = None
def allow_request(self,request,view):
#获取用户ip作为唯一的标示
remote_addr = request.META.get(‘REMOTE_ADDR‘)
# 获取当前访问的时刻
ctime = time.time()
# 这是用户第一次访问,将其进行记录，并且返回True，允许继续访问
if remote_addr not in VISIT_RECORD:
VISIT_RECORD[remote_addr] = [ctime,]
return True
# 如果不是第一次访问，获取所有的记录
history = VISIT_RECORD.get(remote_addr)
self.history = history
# 判断最开始的时刻与现在的时刻的差值是否在规定的时间范围内，比如在60s内，如果不在，
# 可以去除最开始的时刻记录
while history and history[-1] < ctime - 30:
history.pop()
# 此时列表中的时刻记录都是在规定的时间范围内，判断时刻的个数也就是访问的次数
if len(history) < 3:
history.insert(0,ctime)
return True
def wait(self):
# 还需要等多少秒才能访问
ctime = time.time()
return 60 - (ctime - self.history[-1])
在对应的视图中进行配置：
class BookView(ListAPIView):
throttle_classes = [VisitThrottle,] #配置限流组件
queryset = models.Book.objects.all()
serializer_class = BookModelSerializer
2、限流原理
在rest framework框架中，限流定义为类的列表，只需要全局配置或者局部配置即可。上述限流的原理就是以客户端的唯一标示作为键，以访问的时刻形成的列表作为值形成的字典，然后通过对字典进行操作：
{
http://127.0.0.1:8020/ :[11:43:30,11:42:22,11:42:20,11:42:09]
}
如上面的字典所示，后面的访问时间放插入到列表的最左侧，加入当前访问时间是11:43::30，那么与最开始访问时间11:42:09进行做差，然后与规定时间30s进行比较，如果不在30s内，那么就去除最左边的记录，同理使用while循环依次比较，最后在规定时间范围内的记录：
{
http://127.0.0.1:8020/ :[11:43:30,]
}
再计算访问次数，也就是列表的个数，显然如果列表的个数小于3可以继续访问，否则不可以。
上面使用全局变量来进行记录，当然也是可以使用缓存来进行记录的存储，需要使用django的缓存API，from django.core.cache import cache，导入这个API后就可以使用set和get方法，设置和获取cache中存储的对象，只需要在操作全局变量除进行替换即可：
from django.core.cache import cache as default_cache
import time
class VisitThrottle(object):
cache = default_cache
def allow_request(self,request,view):
...
...
# 这是用户第一次访问,将其进行记录，并且返回True，允许继续访问
if not self.cache.get(remote_addr):
self.cache.set(remote_addr,[ctime,])
return True
# 如果不是第一次访问，获取所有的记录
history = self.cache.get(remote_addr)
self.history = history
...
...
rest framework的限流组件就是基于cache来完成的。
上述的wait方法表示还需要多长时间可以进行访问这个API，对客户端的提示：
{
"detail": "Request was throttled. Expected available in 56 seconds."
}
二、内置限流
在rest framework中已经有一些限流的API可以使用：
1、SimpleRateThrottle
class SimpleRateThrottle(BaseThrottle):
"""
A simple cache implementation, that only requires `.get_cache_key()`
to be overridden.
The rate (requests / seconds) is set by a `rate` attribute on the View
class. The attribute is a string of the form ‘number_of_requests/period‘.
Period should be one of: (‘s‘, ‘sec‘, ‘m‘, ‘min‘, ‘h‘, ‘hour‘, ‘d‘, ‘day‘)
Previous request information used for throttling is stored in the cache.
"""
cache = default_cache
timer = time.time
cache_format = ‘throttle_%(scope)s_%(ident)s‘
scope = None
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
def __init__(self):
if not getattr(self, ‘rate‘, None):
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
def get_cache_key(self, request, view):
"""
Should return a unique cache-key which can be used for throttling.
Must be overridden.
May return `None` if the request should not be throttled.
"""
raise NotImplementedError(‘.get_cache_key() must be overridden‘)
def get_rate(self):
"""
Determine the string representation of the allowed request rate.
"""
if not getattr(self, ‘scope‘, None):
msg = ("You must set either `.scope` or `.rate` for ‘%s‘ throttle" %
self.__class__.__name__)
raise ImproperlyConfigured(msg)
try:
return self.THROTTLE_RATES[self.scope]
except KeyError:
msg = "No default throttle rate set for ‘%s‘ scope" % self.scope
raise ImproperlyConfigured(msg)
def parse_rate(self, rate):
"""
Given the request rate string, return a two tuple of:
<allowed number of requests>, <period of time in seconds>
"""
if rate is None:
return (None, None)
num, period = rate.split(‘/‘)
num_requests = int(num)
duration = {‘s‘: 1, ‘m‘: 60, ‘h‘: 3600, ‘d‘: 86400}[period[0]]
return (num_requests, duration)
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()
# Drop any requests from the history which have now passed the
# throttle duration
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()
def throttle_success(self):
"""
Inserts the current request‘s timestamp along with the key
into the cache.
"""
self.history.insert(0, self.now)
self.cache.set(self.key, self.history, self.duration)
return True
def throttle_failure(self):
"""
Called when a request to the API has failed due to throttling.
"""
return False
def wait(self):
"""
Returns the recommended next request time in seconds.
"""
if self.history:
remaining_duration = self.duration - (self.now - self.history[-1])
else:
remaining_duration = self.duration
available_requests = self.num_requests - len(self.history) + 1
if available_requests <= 0:
return None
return remaining_duration / float(available_requests)
SimpleRateThrottle
如果需要借助这个API来实现功能，自己也需要进行一些配置：
- 继承SimpleRateThrottle
自己定义的限流类需要继承SimpleRateThrottle：
class VisitThrottle(SimpleRateThrottle):
...
- 设置scope
class VisitThrottle(SimpleRateThrottle):
scope = ‘book‘
...
在自定义类中设置scope参数，并且还需要在settings中配置DEFAULT_THROTTLE_RATES
REST_FRAMEWORK = {
"DEFAULT_THROTTLE_RATES": {
"book": ‘6/m‘, #每分钟访问6次
}
- 重写get_cache_key方法
class VisitThrottle(SimpleRateThrottle):
scope = ‘book‘
def get_cache_key(self,request,view):
"""
获取访问的标示，比如以ip作为标示
:param request:
:param view:
:return:
"""
remote_addr = request.META.get(‘REMOTE_ADDR‘)
return remote_addr
获取访问的唯一标示ip，当然SimpleRateThrottle继承了BaseThrottle，在BaseThrottle中有获取ip的方法，只需要调用即可。
class VisitThrottle(SimpleRateThrottle):
scope = ‘book‘
def get_cache_key(self,request,view):
return self.get_ident(request)
- 局部配置
只需要在对应的视图中添加对应限流类的列表即可：
class BookView(ListAPIView):
...
throttle_classes = [VisitThrottle,] #配置节流组件
...
- 全局配置
当然也可以在settings中进行全局配置：
REST_FRAMEWORK = {
"DEFAULT_THROTTLE_CLASSES":["app01.utils.throttle.VisitThrottle"],
"DEFAULT_THROTTLE_RATES": {
"book": ‘6/m‘,
}
这样也就完成了相对应的功能，另外内部还提供了其它的API可以使用。
2、AnonRateThrottle
class AnonRateThrottle(SimpleRateThrottle):
"""
Limits the rate of API calls that may be made by a anonymous users.
The IP address of the request will be used as the unique cache key.
"""
scope = ‘anon‘
def get_cache_key(self, request, view):
if request.user.is_authenticated:
return None # Only throttle unauthenticated requests.
return self.cache_format % {
‘scope‘: self.scope,
‘ident‘: self.get_ident(request)
}
AnonRateThrottle
限制未认证的用户。通过传入请求的 IP 地址生成一个唯一的密钥来进行限制。
3、UserRateThrottle
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:
ident = self.get_ident(request)
return self.cache_format % {
‘scope‘: self.scope,
‘ident‘: ident
}
UserRateThrottle
通过 API 将用户请求限制为给定的请求频率。用户标识用于生成一个唯一的密钥来加以限制。未经身份验证的请求将回退到使用传入请求的 IP 地址生成一个唯一的密钥来进行
限制。
4、ScopedRateThrottle
class ScopedRateThrottle(SimpleRateThrottle):
"""
Limits the rate of API calls by different amounts for various parts of
the API. Any view that has the `throttle_scope` property set will be
throttled. The unique cache key will be generated by concatenating the
user id of the request, and the scope of the view being accessed.
"""
scope_attr = ‘throttle_scope‘
def __init__(self):
# Override the usual SimpleRateThrottle, because we can‘t determine
# the rate until called by the view.
pass
def allow_request(self, request, view):
# We can only determine the scope once we‘re called by the view.
self.scope = getattr(view, self.scope_attr, None)
# If a view does not have a `throttle_scope` always allow the request
if not self.scope:
return True
# Determine the allowed request rate as we normally would during
# the `__init__` call.
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
# We can now proceed as normal.
return super(ScopedRateThrottle, self).allow_request(request, view)
def get_cache_key(self, request, view):
"""
If `view.throttle_scope` is not set, don‘t apply this throttle.
Otherwise generate the unique cache key by concatenating the user id
with the ‘.throttle_scope` property of the view.
"""
if request.user.is_authenticated:
ident = request.user.pk
else:
ident = self.get_ident(request)
return self.cache_format % {
‘scope‘: self.scope,
‘ident‘: ident
}
ScopedRateThrottle
可用于限制对 API 特定部分的访问。只有当正在访问的视图包含 .throttle_scope 属性时才会应用此限制。然后通过将请求的 “范围” 与唯一的用户标识或 IP 地址连接起来形成唯一的限流密钥
三、源码剖析
限流组件和权限组件、认证组件等类似，还是从路由对应的视图函数的as_view方法着手，可以看到最终走到的还是APIView的dispatch方法。
1、dispatch
def dispatch(self, request, *args, **kwargs):
"""
`.dispatch()` is pretty much the same as Django‘s regular dispatch,
but with extra hooks for startup, finalize, and exception handling.
"""
self.args = args
self.kwargs = kwargs
#rest-framework重构request对象
request = self.initialize_request(request, *args, **kwargs)
self.request = request
self.headers = self.default_response_headers # deprecate?
try:
self.initial(request, *args, **kwargs)
# Get the appropriate handler method
#这里和CBV一样进行方法的分发
if request.method.lower() in self.http_method_names:
handler = getattr(self, request.method.lower(),
self.http_method_not_allowed)
else:
handler = self.http_method_not_allowed
response = handler(request, *args, **kwargs)
except Exception as exc:
response = self.handle_exception(exc)
self.response = self.finalize_response(request, response, *args, **kwargs)
return self.response
这里的dispatch方法是APIView中的dispatch方法，在这里对原先的request进行了重构，以及通过self.initial(request, *args, **kwargs)加入了限流组件。
2、initial
def initial(self, request, *args, **kwargs):
"""
Runs anything that needs to occur prior to calling the method handler.
"""
self.format_kwarg = self.get_format_suffix(**kwargs)
# Perform content negotiation and store the accepted info on the request
neg = self.perform_content_negotiation(request)
request.accepted_renderer, request.accepted_media_type = neg
# Determine the API version, if versioning is in use.
version, scheme = self.determine_version(request, *args, **kwargs)
request.version, request.versioning_scheme = version, scheme
# Ensure that the incoming request is permitted
self.perform_authentication(request) #进行认证
self.check_permissions(request) #权限校验
self.check_throttles(request) #限流组件
3、check_throttles
def check_throttles(self, request):
"""
Check if request should be throttled.
Raises an appropriate exception if the request is throttled.
"""
for throttle in self.get_throttles():
if not throttle.allow_request(request, self):
self.throttled(request, throttle.wait())
def get_throttles(self):
"""
Instantiates and returns the list of throttles that this view uses.
"""
return [throttle() for throttle in self.throttle_classes]
get_throttles
可以看到循环的是视图中配置的限流类的列表，而且显然每一个限流类都必须要有allow_request和wait方法，如果allow_request返回的False就是说明已经限制访问了，执行self.throttled(request, throttle.wait())。
4、throttled
def throttled(self, request, wait):
"""
If request is throttled, determine what kind of exception to raise.
"""
raise exceptions.Throttled(wait)
也就是如果已经限流了，就会抛出异常，给客户端限流提示。
详情参考：https://q1mi.github.io/Django-REST-framework-documentation/api-guide/throttling/#anonratethrottle
原文地址：https://www.cnblogs.com/shenjianping/p/11494092.html
时间： 2024-10-04 01:51:37