《Redis深度历险》五（限流）

限流Demo

用zset去做限流

client = redis.StrictRedis()

def is_action_allowed(user_id,action_key,period,max_count):
	key = 'hist:%s:%s'%(user_id,action_key)
	now_ts = int(time.time()*1000)
	with client.pipeline() as pipe:
		# 记录行为
		pipe.zadd(key,now_ts,now_ts) # value 和 score 都使用毫秒时间戳
		# 移除时间窗口之前的行为记录，剩下的都是时间窗口内的 pipe.zremrangebyscore(key, 0, now_ts - period * 1000)
		# 获取窗口内的行为数量
		pipe.zcard(key)
		# 设置 zset 过期时间，避免冷用户持续占用内存 
		# 过期时间应该等于时间窗口的长度，再多宽限 1s pipe.expire(key, period + 1)
		pipe.expire(key, period + 1)
	# 比较数量是否超标
	return current_count <= max_count

缺点：比如限定60s不得超过100w次这样的参数，不适合做这样的限流，因为会消耗大量的存储空间。

Funnels限流

class Funnel(object):
    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 = "%s:%s"%(user_id,action_key)
        funnel = funnels.get(key)
    
        if not funnel:
            funnel = Funnel(capacity,leaking_rate)
            funnels[key] = funnel
        return funnel.watering(1)
    
    for i in range(20):
        print(is_action_allowed("jack","reply",15,0.5))

但是有个问题，我们无法保证整个过程的原子性。从 hash 结构中取值，然后在内存里运算，再回填到 hash 结构，这三个过程无法原子化，意味着需要进行适当的加锁控制。而一旦加锁，就意味着会有加锁失败，加锁失败就需要选择重试或者放弃。

如果重试的话，就会导致性能下降。如果放弃的话，就会影响用户体验。同时，代码的复杂度也跟着升高很多。

Redis-Cell

cl.throttle

cl.throttle jack:reply 15 30 60 1
    					key
    									capacity(漏斗容量15)
    											(漏水速率30/60)
    														可选参数（默认值1）

标识允许jack:reply行为每60s最多30次，漏斗的初始容量为15，也就是一开始连续回复15个帖子，才开始受漏水速率的影响。

cl.throttle jack:reply 15 30 60
0 # 0标识允许，1表示拒绝
15 # 漏斗容量
14 # 漏斗剩余空间
-1 # 如果拒绝了，需要多久重试（单位秒）
2  # 多长时间后，漏斗完全空出来

在执行限流指令时，如果被拒绝了，就需要丢弃或重试。cl.throttle 指令考虑的非常周到，连重试时间都帮你算好了，直接取返回结果数组的第四个值进行 sleep 即可，如果不想阻塞线程，也可以异步定时任务来重试。