Open codcodog opened 6 years ago
在对接 IB 接口的时候,它们接口有一个严格的规定:
IB
用户 1s 内的请求不能超过50个
在对接的时候,采用 Python 的 协程 + 事件循环,也就是所谓的异步,每秒向 TWS 发送 50个请求,在请求 每天 级别的数据时,接口可以正常工作。
Python
协程
事件循环
异步
TWS
每天
但,在请求 每5分钟 级别的数据时,接口不能正常工作:当请求达到一定的数量时就会触发接口的规定
每5分钟
ERROR:ib_insync.wrapper:Error 100, reqId 20: Max rate of messages per second has been exceeded:max=50 rec=85 (3)
很好奇 IB 是如何实现用户 1s 内的请求不能超过50个的?
在高并发系统下,有三把利器:缓存,降级,限流
缓存
将常用数据缓存起来,减少数据库或者磁盘 IO
降级
保护核心系统,降低非核心业务请求响应
限流
在某一个时间窗口内对请求进行限速,保护系统
这里主要针对限流进行介绍,常见的限流算法有:
计数器法 就是为客户端的请求设置一个计数器 Counter, 记录客户端的请求量(QPS)
计数器法
Counter
对于 1s 内用户的请求不能超过50个,可以这样简单实现: 一开始,我们为用户设置一个计数器 Counter,每当一个请求过来的时候,Counter 加1. 如果该请求与第一个请求的间隔时间在1s之内,但请求数 Counter 大于50的时候,则说明用户 1s 内的请求数过多了. 如果该请求与第一个请求的间隔时间大于1s,则重置 Counter.
import time class Counter: def __init__(self): ''' 初始化计数器 @param limit 单位时间请求最大值 @param interval 单位时间:秒 ''' self.req = 0 self.limit = 50 self.interval = 1 self.timestamp = time.time() def is_out_rate(self): ''' 是否超过限制 ''' now = time.time() delta = now - self.timestamp if delta <= self.interval: self.req += 1 return self.req > self.limit else: # 超时重置 self.timestamp = now self.req = 1 return False
计数器法 存在临界问题: 设置 1s 内用户的请求不超过50个,如果将粒度设置为 ms,则期望用户的请求:0.05 r/ms
ms
假设有一个恶意用户,在 1s 瞬间发送了50个请求,在 1.001s 又瞬间发送了50个请求,则在 0.001s(1ms) 发送了100个请求,也就是 100 r/ms 假设有一个恶意用户,在0s发送了1个请求,在1s瞬间发送了49个请求,在 1.001s 又瞬间发送了50个请求,则在 0.001s(1ms) 发送了99个请求,也就是 99 r/ms 这样利用时间窗口的重置节点突发请求,就可以瞬间突破我们的速率限制.
对于设置秒级的Counter 来说,其实精度已经很高了,这里是为了说明需要,采用毫秒进行举例 日常开发中,如果是分钟,小时这种低精度的话,则需要考虑临界问题
临界问题 参考:滑动窗口
临界问题
漏桶算法:水(请求)流入到漏桶里,漏桶以一定的速度流出,当水流入速度过大则会直接溢出.
漏桶算法
漏桶算法 天生就限制了请求的速度,并且不会存在临界问题.
首先有一个固定容量的桶,有水(请求)流进来和流出去。对于水(请求)流进来的速度不做限制,但固定水(请求)流出的速率,当桶满了之后,多余的水(请求)则溢出(拒绝/排队)。
import time class LeackyBucket: def __init__(self, capacity, out_rate): ''' 初始化漏桶 @param capacity 桶的容量 @param out_rate 流出的速率 @param water 当前的水的容量 ''' self.capacity = float(capacity) self.out_rate = float(out_rate) self._water = 0 self.timestamp = time.time() def consume(self, water): ''' 往漏桶加水 ''' if self._water + water > self.capacity: return False self._water += water return True @property def water(self): ''' 查看当前漏桶中的水量 ''' if self._water < self.capacity: now = time.time() delta = self.out_rate * (now - self.timestamp) self._water = self._water - delta self.timestamp = now return self._water else: return self.capacity if __name__ == '__main__': bucket = LeackyBucket(50, 50) print('漏桶的容量:%.2f,当前桶的水量:%.2f,流出的速度是:%.2f r/s' % (bucket.capacity, bucket.water, bucket.out_rate)) print('往桶加入50水量:%s' % bucket.consume(50)) print('当前桶的水量:%.2f' % bucket.water) time.sleep(0.1) print('约0.1s之后,桶的水量是:%.2f' % bucket.water) time.sleep(0.2) print('约0.3s之后,桶的水量是:%.2f' % bucket.water) print('往桶加入40水量:%s' % bucket.consume(40))
[cryven@codcodog ~]$ python leaky_buckey.py 漏桶的容量:50.00,当前桶的水量:-0.00,流出的速度是:50.00 r/s 往桶加入50水量:True 当前桶的水量:50.00 约0.1s之后,桶的水量是:44.99 约0.3s之后,桶的水量是:34.97 往桶加入40水量:False
令牌桶算法:系统以一个恒定的速度往桶里放入令牌,如果请求需要被处理,则需要先从桶里获取一个令牌,若桶里没有令牌时,则拒绝服务.
令牌桶算法
有一个固定容量的桶,用来存放令牌(token)。一开始,桶是空的,令牌(token)以固定的速率 R 往桶里填充,如果桶满了,则多余的令牌被丢弃。 当一个请求过来的时候,首先从桶里获取令牌(token),获取成功后,请求才会被接受;若获取失败,则请求被拒绝,或者排队等待,直到获取令牌(token)成功。
import time class TokenBucket: def __init__(self, tokens, fill_rate): ''' 初始化令牌桶 @param tokens 令牌桶的大小 @param fill_rate 填充令牌的速度,tokens/second ''' self.capacity = float(tokens) self._tokens = float(tokens) self.fill_rate = float(fill_rate) self.timestamp = time.time() def consume(self, tokens): ''' 消费令牌桶中的令牌 桶中有足够的令牌则返回 True 否则返回 False ''' if tokens <= self._tokens: self._tokens -= tokens return True return False @property def tokens(self): ''' 获取 bucket 的 token ''' if self._tokens <= self.capacity: now = time.time() delta = self.fill_rate * (now - self.timestamp) self._tokens = min(self.capacity, self._tokens + delta) self.timestamp = now return self._tokens if __name__ == '__main__': bucket = TokenBucket(50, 50) print('当前桶中的tokens:%.2f,填充速度是:%.2f t/s' % (bucket.tokens, bucket.fill_rate)) print('消耗30个token:%s,当前tokens:%.2f' % (bucket.consume(30), bucket.tokens)) time.sleep(0.1) print('约0.1秒后当前桶中的tokens:%.2f' % bucket.tokens) time.sleep(0.2) print('约0.3秒后当前桶中的tokens:%.2f' % bucket.tokens) print('消耗60个token:%s' % bucket.consume(60)) print('当前桶中的tokens:%.2f' % bucket.tokens)
[cryven@codcodog ~]$ python token_bucket.py 当前桶中的tokens:50.00,填充速度是:50.00 t/s 消耗30个token:True,当前tokens:20.00 约0.1秒后当前桶中的tokens:25.01 约0.3秒后当前桶中的tokens:35.03 消耗60个token:False 当前桶中的tokens:35.04
代码来源:IMPLEMENTATION OF THE TOKEN BUCKET ALGORITHM
计数器 是最简单粗暴的,但当精度低的时候会存在 临界问题,可以使用 滑动窗口 进行调整.
计数器
滑动窗口
漏桶算法 和 令牌桶算法 最大的区别就是 令牌桶算法 允许流量一定程度的突发:如果桶内有50个Token时,那么可以瞬间允许50个请求通过,而 漏桶算法 无法做到.
最后,咨询和了解了下,感觉不是我们请求频率的问题,猜测 IB 做了类似反爬虫的策略,毕竟2800多个股短时间内请求分钟颗粒的数据,估计当我们请求的量达到一定的程度时,IB 就开始拒绝我们的请求了,毕竟它不是专业的数据提供商.
参考文章 接口限流算法总结
假设有一个恶意用户,在 1s 瞬间发送了50个请求,在 1.001s 又瞬间发送了50个请求,则在 0.001s(1ms) 发送了100个请求,也就是 100 r/ms
应该是按第一次请求开始算时间吧?比如说:0s 的时候发送第一次请求,在0.999s 的时候发送剩余49次。
限流
问题触发的场景
在对接
IB
接口的时候,它们接口有一个严格的规定:在对接的时候,采用
Python
的协程
+事件循环
,也就是所谓的异步
,每秒向TWS
发送 50个请求,在请求每天
级别的数据时,接口可以正常工作。但,在请求
每5分钟
级别的数据时,接口不能正常工作:当请求达到一定的数量时就会触发接口的规定很好奇
IB
是如何实现用户 1s 内的请求不能超过50个的?问题扩展
在高并发系统下,有三把利器:缓存,降级,限流
缓存
降级
限流
这里主要针对限流进行介绍,常见的限流算法有:
计数器算法
计数器法
就是为客户端的请求设置一个计数器Counter
, 记录客户端的请求量(QPS)对于 1s 内用户的请求不能超过50个,可以这样简单实现:
一开始,我们为用户设置一个计数器
Counter
,每当一个请求过来的时候,Counter
加1.如果该请求与第一个请求的间隔时间在1s之内,但请求数
Counter
大于50的时候,则说明用户 1s 内的请求数过多了.如果该请求与第一个请求的间隔时间大于1s,则重置
Counter
.代码实现
存在问题
计数器法
存在临界问题:设置 1s 内用户的请求不超过50个,如果将粒度设置为
ms
,则期望用户的请求:0.05 r/ms假设有一个恶意用户,在 1s 瞬间发送了50个请求,在 1.001s 又瞬间发送了50个请求,则在 0.001s(1ms) 发送了100个请求,也就是 100 r/ms假设有一个恶意用户,在0s发送了1个请求,在1s瞬间发送了49个请求,在 1.001s 又瞬间发送了50个请求,则在 0.001s(1ms) 发送了99个请求,也就是 99 r/ms 这样利用时间窗口的重置节点突发请求,就可以瞬间突破我们的速率限制.
临界问题
参考:滑动窗口漏桶算法
漏桶算法
:水(请求)流入到漏桶里,漏桶以一定的速度流出,当水流入速度过大则会直接溢出.首先有一个固定容量的桶,有水(请求)流进来和流出去。对于水(请求)流进来的速度不做限制,但固定水(请求)流出的速率,当桶满了之后,多余的水(请求)则溢出(拒绝/排队)。
代码实现
令牌桶算法
令牌桶算法
:系统以一个恒定的速度往桶里放入令牌,如果请求需要被处理,则需要先从桶里获取一个令牌,若桶里没有令牌时,则拒绝服务.有一个固定容量的桶,用来存放令牌(token)。一开始,桶是空的,令牌(token)以固定的速率 R 往桶里填充,如果桶满了,则多余的令牌被丢弃。
当一个请求过来的时候,首先从桶里获取令牌(token),获取成功后,请求才会被接受;若获取失败,则请求被拒绝,或者排队等待,直到获取令牌(token)成功。
代码实现
代码来源:IMPLEMENTATION OF THE TOKEN BUCKET ALGORITHM
总结
计数器
是最简单粗暴的,但当精度低的时候会存在临界问题
,可以使用滑动窗口
进行调整.漏桶算法
和令牌桶算法
最大的区别就是令牌桶算法
允许流量一定程度的突发:如果桶内有50个Token时,那么可以瞬间允许50个请求通过,而漏桶算法
无法做到.最后,咨询和了解了下,感觉不是我们请求频率的问题,猜测
IB
做了类似反爬虫的策略,毕竟2800多个股短时间内请求分钟颗粒的数据,估计当我们请求的量达到一定的程度时,IB
就开始拒绝我们的请求了,毕竟它不是专业的数据提供商.参考文章
接口限流算法总结