漏斗限流是最常用的限流方法之一，漏斗流水的速率大于灌水的速率，漏斗就永远装不满，反之水就会溢出。
所以漏斗的剩余空间就代表当前行为可以持续进行的数量，水流出的速率代表系统允许该行为的最大频率。
import time
class Funnel:
def __init__(self, capacity, leaking_rate):
self.capacity = capacity # 漏斗容量
self.leaking_rate = leaking_rate # 流出速率
self.left_quota = capacity # 漏斗剩余空间
self.leaking_ts = time.time() # 上一次漏水时间
def make_space(self):
now_ts = time.time()
delta_ts = now_ts - self.leaking_ts # 距离上一次漏水过去了多久
delta_quota = delta_ts * self.leaking_rate # 腾出的空间
if delta_quota < 1: # 腾的空间过小等待下一次
return
self.left_quota += delta_quota # 增加剩余空间
self.leaking_ts = now_ts # 记录漏水时间
if self.left_quota > self.capacity: # 剩余空间不得高于容量
self.left_quota = self.capacity
def watering(self, quota):
self.make_space()
if self.left_quota >= quota: # 判断剩余空间是否足够
self.left_quota -= quota
return True
return False
funnels = {} # 所有的漏斗
def is_action_allowed(
user_id, action_key, capacity, leaking_rate):
key = '{}:{}'.format(user_id, action_key)
funnel = funnels.get(key)
if not funnel:
funnels[key] = Funnel(capacity, leaking_rate)
return funnel.watering(1)
for i in range(20):
print is_action_allowed('tom', 'reply', 15, 0.5)make_space方法是漏斗算法的核心，其在每次灌水前都会被调用以触发漏水，给漏斗腾出空间，Funnel占据的空间大小不与行为频率成正比，其空间占用是一个常量。
问题是分布式漏斗算法如何实现？
Funnel类其实就是一个高级字典，那么我们可以利用Redis中的hash结构来存储对应字段，灌水时将字段取出进行逻辑运算后再存入hash结构中即可完成一次行为频度的检测。但这有个问题就是整个过程的原子性无法保证，意味着要用锁来控制，但如果加锁失败，就要重试或者放弃，这回导致性能下降和影响用户体验，同时代码复杂度也升高了，此时Redis提供了一个插件，Redis-Cell出现了。
Redis-Cell
Redis 4.0提供了一个限流Redis模块，名称为redis-cell，该模块提供漏斗算法，并提供原子的限流指令。
该模块只有一条指令cl.throttle，其参数和返回值比较复杂。
> cl.throttle tom:reply 14 30 60 1
1) (integer) 0 # 0表示允许，1表示拒绝
2) (integer) 15 # 漏斗容量capacity
3) (integer) 14 # 漏斗剩余空间left_quota
4) (integer) -1 # 如果拒绝了，需要多长时间后再重试，单位秒
5) (integer) 2 # 多长时间后，漏斗完全空出来，单位秒该指令意思为，允许用户tom的reply行为的频率为每60s最多30次，漏斗初始容量为15（因为是从0开始计数，到14为15个），默认每个行为占据的空间为1（可选参数）。
如果被拒绝，取返回数组的第四个值进行sleep即可作为重试时间，也可以异步定时任务来重试。
注意事项
由于Redis-Cell是基于Rust语言写的插件，因此在安装插件前要先安装rust。
具体方法看GitHub。