45.限流Throttling及源码解析

时间:2022-10-25 16:07:10

什么是限流?

  1. 限流类似于权限机制,它也决定是否接受当前请求,用于控制客户端在某段时间内允许向API发出请求的次数,也就是频率
  2. 假设有客户端(比如爬虫程序)短时间发起大量请求,超过了服务器能够处理的能力,将会影响其他用户的正常使用
  3. 为了保证服务的稳定性,并防止接口受到恶意用户的攻击,我们可以对接口进行限流
  4. 又或者可以对未经身份验证的请求设置访问频率,对经过身份验证的请求不限制访问频率
  5. 限流也不止单指限制访问次数的措施,例如付费数据服务的特点访问次数

限流的应用场景

  1. 区分用户场景,比如匿名和已登录,不同权限的用户不同的限流策略
  2. API的不同,根据不同API设置不同的策略
  3. 请求的爆发期和持续期不同的限流策略
  4. 可以同时支持使用多个限流策略
 
 

限流的机制

限流和权限一样,执行视图前会依次检查所有的限流类,全部通过会执行View,任何一个检查失败,会抛出Exceptions.Throttled异常
在settings中,通过 DEFAULT_THROTTLE_CLASSES 设置限流类,通过DEFAULT_THROTTLE_RATES设置限流频率
 
 

DRF提供的两个常用限流类

AnonRateThrottle:对于匿名用户的限流,使用anon设置频率
UserRateThrottle:对于登录用户的限流, 使用user设置频率
 

全局限流类配置

REST_FRAMEWORK = {
    # 全局限流类的配置
    'DEFAULT_THROTTLE_CLASSES': (
        'rest_framework.throttling.AnonRateThrottle',  # 对于匿名用户的限流
        'rest_framework.throttling.UserRateThrottle' #对于登录用户的限流
    ),
    # 限流频率的配置
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day', # 未认证用户一天只许访问100次
        'user': '1000/day' # 认证用户一天可以访问1000次
        }
    }
DEFAULT_THROTTLE_RATES设置限流频率格式 次数/时间单位
  • second: 按秒设置频率次数
  • minute:按分钟设置频率次数
  • hour:按小时设置频率次数
  • day: 按天设置频率次数
 

视图级别限流类配置

#导入限流模块
from rest_framework import throttling 


class getInfoList(ModelViewSet):
    # 通过throttle_classes 设置该视图的限流类
    # 视图指定会覆盖settings设置的全局限流
    throttle_classes = (throttling.UserRateThrottle,)

    def infoList(self):
        ...
 
 

识别请求的客户端

我们既然要对请求进行限流,那么肯定要失识 别是谁发来的请求,然后进行对应的措施,不然无法确定请求者身份,那么就无法得知是不是需要限制的请求,常见的方法有三种
  1. drf利用http报头的 x-forwarded-for 或者wsgi中的remote-addr变量来唯一标识客户端的IP地址
  2. 如果 存在x-forwarded-for 属性,则使用x-forwarded-for ,否则使用remote-addr
  3. 可以使用request.user的属性来标识请求,比如使用request.user.id 来标记唯一请求
  4. 使用IP地址对客户端请求进行限流,需要考虑使用伪造代理IP请求的情况
 
 

throttling源码解析

throttling源码一共有五个类
  1. BaseThrottle: 限流基类
  2. SimpleRateThrottle:频率校验类
  3. AnonRateThrottle:匿名用户限流
  4. UserRateThrottle:认证用户限流
  5. ScopedRateThrottle:api视图级别的限流
 

BaseThrottle限流基类

没有去具体实现某些功能,跟权限类基类似,只是提供了占位方法
class BaseThrottle:

    # allow_request源码并没有直接实现功能,只是写好了方法占位,待后续继承实现
    # 该方法主要是处理是否允许请求通过
    # 如果后续继承基类实现该方法,允许请求通过返回True,不允许请求通过返回False
    def allow_request(self, request, view):
      
        raise NotImplementedError('.allow_request() must be overridden')

    # 获取IP地址
    def get_ident(self, request):
        # 获取请求头中真实IP地址
        xff = request.META.get('HTTP_X_FORWARDED_FOR')
        # 获取代理IP地址
        remote_addr = request.META.get('REMOTE_ADDR')
        # 获取设置的允许的最大代理数,默认不设置为None
        num_proxies = api_settings.NUM_PROXIES
       # 如果num_proxies不是None,说明设置了该值
        if num_proxies is not None:
            # 如果设置为0,或者 xff没有值
            if num_proxies == 0 or xff is None:
                # 返回代理IP地址
                return remote_addr
            #使用代理IP的话会有多个地址,使用逗号分割成一个list
            addrs = xff.split(',')
            ''' 
            通过min函数,拿到允许的代理数和IP地址长度最小的值,使用-变成负数
            在addrs列表中通过该下标取对应值
            '''
            client_addr = addrs[-min(num_proxies, len(addrs))]
            return client_addr.strip()
        # 如果没有设置允许的代理数 并且xff有值则直接返回,否则返回remote_addr
        return ''.join(xff.split()) if xff else remote_addr

    # 等待时间,告诉客户端被限流,等待多久可以访问
    # 后续继承实现,可选
    def wait(self):
    
        return None

SimpleRateThrottle

频率控制类,继承了BaseThrottle,添加和重写了一些方法,重点是添加了get_cache_key 方法,但必须自己实现该方法
class SimpleRateThrottle(BaseThrottle):
    # 限流需要用到缓存,使用drf默认的缓存
    # 如果其他继承类想修改缓存机制,cache = caches['缓存名'] 进行修改
    cache = default_cache
    # time.time方法,但是并没有()进行实例调用
    # 类似计时器功能,在这里留好,后续调用
    timer = time.time  
    # 缓存设置,字符串格式化方法后续传参使用
    cache_format = 'throttle_%(scope)s_%(ident)s'
    # scope默认没有设置,该值是DEFAULT_THROTTLE_RATES中对应限流类的key
    scope = None
    # 限流频率默认的配置值
    THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES

    def __init__(self):
        if not getattr(self, 'rate', None):
            # 从下面get_rate()方法获取访问频率限制的参数
            self.rate = self.get_rate()
            
        # 通过self.parse_rate方法获取限制的频率及持续时间赋值给num_requests
        self.num_requests, self.duration = self.parse_rate(self.rate)
    # 获取当前请求的标识
    def get_cache_key(self, request, view):
        raise NotImplementedError('.get_cache_key() must be overridden')
        
    # 获取settings频率设置限流类对应的key
    def get_rate(self):
       # 如果没有scope,抛出异常
        if not getattr(self, 'scope', None):
            msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
                   self.__class__.__name__)
            raise ImproperlyConfigured(msg)

        try:
            # 从 self.THROTTLE_RATES 中获取设置的scope
            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):
         # 如果没有设置频率限制,直接返回None
        if rate is None:
            return (None, None)
        # 在settings设置频率我们使用 num/type 设置值
        # 字符串使用/分割 ,获取两个对应的值
        num, period = rate.split('/')
        num_requests = int(num)
        #settings中设置时间单位以天为单位可以是day也可以是d
        # period[0]获取第一个字符为key,以秒为单位换算,秒就1,分就是60,天就是86400 
        # 如果需要扩展月、年等时间,可以扩展源码
        duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
        # 返回频率和持续时间
        return (num_requests, duration)

    # 是否允许请求通过,运行返回True,否则返回False
    def allow_request(self, request, view):
        # 如果没有设置限流频率,直接返回True
        if self.rate is None:
            return True
        # 获取用户标识赋值给self.key
        self.key = self.get_cache_key(request, view)
        # 没有用户标识直接返回True
        if self.key is None:
            return True
        # 获取历史访问时间戳
        self.history = self.cache.get(self.key, [])
        # 获取当前时间戳
        self.now = self.timer()

        # while循环,如果历史访问时间戳有值,拿到历史时间戳[-1]的数据,如果小于等于当前时间戳减去持续时间,弹出最后一个时间戳
        # 当前时间-持续时间,就相当于需要限制的时间区间,如果历史时间戳小于等于该区间,才不会继续pop
        while self.history and self.history[-1] <= self.now - self.duration:
            self.history.pop()
        # 如果历史访问时间戳的列表长度大于等于我们设置频率数量,说明到了频率上限
        if len(self.history) >= self.num_requests:
            # 返回self.throttle_failure 对应False
            return self.throttle_failure()
        # 返回self.throttle_success 对应True
        return self.throttle_success()
    # 频率未到达上限时返回该方法
    def throttle_success(self):
        # 在历史请求时间戳列表,将当前时间插入该列表  
        self.history.insert(0, self.now)
        # 更新缓存内容
        self.cache.set(self.key, self.history, self.duration)
        # 返回True
        return True
    # 频率到达上限时返回该方法
    def throttle_failure(self):
       
        return False
    # 返回还需要多长时间可以进行下一次请求,可选方法
    def wait(self):
     
        if self.history:
            # 如果历史请求时间戳有值,剩余时间等于持续时间减去(当前时间-第一次请求)
            remaining_duration = self.duration - (self.now - self.history[-1])
        else:
            # 剩余的时间等于持续时间
            remaining_duration = self.duration
        # 允许请求的次数 等于 允许的次数-已请求的次数+1
        available_requests = self.num_requests - len(self.history) + 1
        # 如果允许请求的次数小于等于0,返回None
        if available_requests <= 0:
            return None

        return remaining_duration / float(available_requests)

AnonRateThrottle

匿名限流类:继承了SimpleRateThrottle,重写了 get_cache_key 方法
AnonRateThrottle 只会限制未经身份验证的用户。传入的请求的IP地址用于生成一个唯一的密钥。
允许的请求频率由以下各项之一确定(按优先顺序):
  1. 类的 rate 属性,可以通过继承 AnonRateThrottle 并设置该属性来修改这个值,优先级高
  2. settings配置文件中 DEFAULT_THROTTLE_RATES['anon'] 配置项的值。优先级低
  3. anonratetrottle 适用于想限制来自未知用户的请求频率的情况
class AnonRateThrottle(SimpleRateThrottle):
    # 设置频率控制的key为anon
    scope = 'anon'
    # 重写get_cache_key方法
    def get_cache_key(self, request, view):
        # 如果请求用户是经过认证的用户,不需要进行限流,直接返回None
        if request.user.is_authenticated:
            return None  
        # 如果用户是未经认证的用户,将该类的scope和 用户的IP地址传入SimpleRateThrottle的self.cache_format类属性
        return self.cache_format % {
            'scope': self.scope,
            'ident': self.get_ident(request)
        }

UserRateThrottle

认证用户限流类:继承了SimpleRateThrottle,仅仅是重写了 get_cache_key 方法
UserRateThrottle 用于限制已认证的用户在整个API中的请求频率。用户ID用于生成唯一的密钥。未经身份验证的请求将使用传入的请求的IP地址生成一个唯一的密钥
允许的请求频率由以下各项之一确定(按优先顺序):
  1. 类的 rate 属性,可以通过继承 UserRateThrottle 并设置该属性来修改这个值,优先级高
  2. settings配置文件中 DEFAULT_THROTTLE_RATES['user'] 配置项的值。优先级低

   # 设置频率控制的key位anon
    scope = 'user'
    # 重写get_cache_key方法
    def get_cache_key(self, request, view):
        if request.user.is_authenticated:
            # 如果请求用户是认证用户,设置用户的唯一标识赋值给ident
            ident = request.user.pk
        else:
            #如果请求用户是非认证用户,通过get_ident获取请求ip赋值给ident
            ident = self.get_ident(request)
        # 设置SimpleRateThrottle中self.cache_format的值
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }

ScopedRateThrottle

用户对于每个视图的访问频次:继承了SimpleRateThrottle,重写了 get_cache_key 和allow_request 方法
ScopedRateThrottle 类用于限制对APIs特定部分的访问,也就是视图级别的限流,不是全局性的
只有当正在访问的视图包含 throttle_scope 属性时,才会应用此限制。然后,通过将视图的“scope”属性值与唯一的用户ID或IP地址连接,生成唯一的密钥。
允许的请求频率由 scope 属性的值在 DEFAULT_THROTTLE_RATES 中的设置确定
class ScopedRateThrottle(SimpleRateThrottle):
    
    scope_attr = 'throttle_scope'

    def __init__(self):

        pass

    def allow_request(self, request, view):
        #  从view获取self.scope_attr赋值给scope,如果view中没有指定,设置为None
        self.scope = getattr(view, self.scope_attr, None)
        # 如果没有设置scope,直接返回True
        if not self.scope:
            return True
        # 获取settings频率设置限流类对应的key  
        self.rate = self.get_rate()
        # 获取频率限制、持续时长
        self.num_requests, self.duration = self.parse_rate(self.rate)
        # 调用父类的allow_request 返回对应的结果
        return super().allow_request(request, view)
    # 获取用户唯一标识
    def get_cache_key(self, request, view):
         # 如果是认证用户 ident=用户唯一标识
        if request.user.is_authenticated:
            ident = request.user.pk
        else:
            # 非认证用户返回请求的ip
            ident = self.get_ident(request)
        # 设置父类的类属性
        return self.cache_format % {
            'scope': self.scope,
            'ident': ident
        }
 

自定义限流类

上面源码的类,我们一般使用的是后三个,如果源码提供的限流类无法满足我们的需求,我们可以写自定义的限流类

自定义限流类的步骤:

  1. 继承BaseThrottle类或者根据场景继承其他限流类
  2. 实现allow_request方法,如果请求被允许,那么返回True,否则返回False
  3. wait方法,是否实现根据自己场景
  4. 获取唯一标识的方法可以使用源码*的,也可以自定义
 
 

场景案例1

假设我们的请求需要同时进行多个认证用户的限流措施,比如每小时限制100次,同时每天限制1000次
# 每小时的限流类
class UserHourRateThrottle(UserRateThrottle): 
    scope = 'userHour' 
# 每天的限流类
class UserDayRateThrottle(UserRateThrottle):
    scope = 'userDay' 
# settings中进行配置
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': (
    # 配置我们自定义的限流类或者再view中进行局部的配置
    'testApi.throttles.UserHourRateThrottle', 
    'testApi.throttles.UserDayRateThrottle'
    ),
    
'DEFAULT_THROTTLE_RATES': {
    'userHour': '100/hour', # 每小时最多100次
    'userDay': '1000/day' # 每天最多100次
}
}

场景案例2

随机限制
import random
class RandomRateThrottle(throttling.BaseThrottle):
    def allow_request(self, request, view):
        # 如果随机的数字 不等于1,返回True,否则返回False
        return random.randint(1, 10) != 1
# 之后在settings进行配置或者局部配置
45.限流Throttling及源码解析