博客

漏斗限流

漏斗思想：

漏洞的容量是有限的,如果将漏嘴堵住,然后一直往里面灌水,它就会变满,直至再也装不进去。如果将漏嘴放开,水就会往下流,流走一部分之后,就又可以继续往里面灌水。如果漏嘴流水的速率大于灌水的速率,那么漏斗永远都装不满。如果漏嘴流水速率小于灌水的速率,那么一旦漏斗满了,灌水就需要暂停并等待漏斗腾空。



所以,漏斗的剩余空间就代表着当前行为可以持续进行的数量,漏嘴的流水速率代表着系统允许该行为的最大频率。

漏斗限流算法

# coding: utf8
import time
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 = {} # 所有的漏斗
# capacity 漏斗容量
# leaking_rate 漏嘴流水速率 quota/s
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('laoqian', 'reply', 15, 0.5)

public class FunnelRateLimiter {
    static class Funnel {
        // 容量
        int capacity;
        // 漏嘴流水速率
        float leakingRate;
        // 漏斗剩余空间
        int leftQuota;
        // 上一次漏水时间
        long leakingTs;
        public Funnel(int capacity, float leakingRate) {
            this.capacity = capacity;
            this.leakingRate = leakingRate;
            this.leftQuota = capacity;
            this.leakingTs = System.currentTimeMillis();
        }
        void makeSpace() {
            long nowTs = System.currentTimeMillis();
            long deltaTs = nowTs - leakingTs;
            int deltaQuota = (int) (deltaTs * leakingRate);
            if (deltaQuota < 0) { // 间隔时间太长,整数数字过大溢出
                this.leftQuota = capacity;
                this.leakingTs = nowTs;
                return;
            }
            if (deltaQuota < 1) { // 腾出空间太小,最小单位是 1
                return;
            }
            this.leftQuota += deltaQuota;
            this.leakingTs = nowTs;
            if (this.leftQuota > this.capacity) {
                this.leftQuota = this.capacity;
            }
        }
        boolean watering(int quota) {
            makeSpace();
            if (this.leftQuota >= quota) {
                this.leftQuota -= quota;
                return true;
            }
            return false;
        }
    }
    private Map<String, Funnel> funnels = new HashMap<>();
    public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {
        String key = String.format("%s:%s", userId, actionKey);
        Funnel funnel = funnels.get(key);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate);
            funnels.put(key, funnel);
        }
        return funnel.watering(1); // 需要 1 个 quota
    }
}

以上是代码实现漏斗限流的思想，但是在分布式环境中，用Redis的会更多。

Redis-Cell

Redis 4.0 提供了一个限流 Redis 模块,它叫 redis-cell。该模块也使用了漏斗算法,并提供了原子的限流指令。有了这个模块,限流问题就非常简单了

该模块只有 1 条指令 cl.throttle,它的参数和返回值都略显复杂,接下来让我们来看看这个指令具体该如何使用。

• laoqian 就是 key值
• 15 是令牌的初始容量，初始容量是该数加一，也就是16，初始化时，令牌是满的
• 30 60 就是速率，意思是60秒内可以请求30次
• 1 表示执行命令时取走的令牌数量，默认是 1

127.0.0.1:6379> cl.throttle user123 15 30 60 1
1) (integer) 0  #0表示允许，1表示拒绝
2) (integer) 16 #漏斗的容量是 16
3) (integer) 15 #漏斗中还剩余的数量，本次执行取走了一个，所以还有15个
4) (integer) -1 #如果拒绝了需要等待的时间，单位是秒
5) (integer) 2  #表示漏斗装满需要等待的时间

127.0.0.1:6379> cl.throttle user123 15 30 60 4
1) (integer) 0
2) (integer) 16
3) (integer) 12
4) (integer) -1
5) (integer) 8
127.0.0.1:6379> cl.throttle user123 15 30 60 4
1) (integer) 0
2) (integer) 16
3) (integer) 8
4) (integer) -1
5) (integer) 14
127.0.0.1:6379> cl.throttle user123 15 30 60 4
1) (integer) 0
2) (integer) 16
3) (integer) 5
4) (integer) -1
5) (integer) 20
127.0.0.1:6379> cl.throttle user123 15 30 60 4
1) (integer) 0
2) (integer) 16
3) (integer) 2
4) (integer) -1
5) (integer) 27
127.0.0.1:6379> cl.throttle user123 15 30 60 4
1) (integer) 1
2) (integer) 16
3) (integer) 2
4) (integer) 2
5) (integer) 26
127.0.0.1:6379> cl.throttle user123 15 30 60 4
1) (integer) 0
2) (integer) 16
3) (integer) 0
4) (integer) -1
5) (integer) 31
127.0.0.1:6379> cl.throttle user123 15 30 60 17
1) (integer) 1
2) (integer) 16
3) (integer) 1
4) (integer) -1
5) (integer) 28
127.0.0.1:6379> cl.throttle user123 15 30 60 17
1) (integer) 1
2) (integer) 16
3) (integer) 16
4) (integer) -1
5) (integer) 0