上一篇中写了一个基于zset的简单限流策略，但是很明显的，它的缺陷很明显，因此本篇中将介绍一种更为常用的漏斗限流。

 

从命名可以看出来，该限流策略的原理就是一个漏斗一样，漏斗将会满足以下三个条件：

　　如果将漏斗口堵上，那么在里面灌满水之后就无法继续装进去。

　　将漏斗口放开，如果单位时间内灌进去的水少于流出去的水，那么漏斗将永远不会满。

　　将漏斗口放开，如果单位时间内灌进去的水多于流出去的水，那么漏斗必将在之后的某个时间被灌满，如果想要继续灌水，就得等待一段时间。

漏斗的剩余空间就代表着可以可以处理任务的一个数量，单位时间内流出去的水就相当于系统在这段时间内可以处理的任务数量。

 

首先是一个单机版本的漏斗算法：

 1 import time
 2 
 3 
 4 class Funnel:
 5     def __init__(self, capacity, leaking_rate):
 6         self.capacity = capacity  # 漏斗容量
 7         self.leaking_rate = leaking_rate  # 漏斗的流水速率
 8         self.left_quota = capacity  # 漏斗的剩余空间，初始化的时候漏斗是空的，所以剩余空间就是漏斗容量
 9         self.leaking_ts = time.time()  # 上一次漏水时间
10 
11     def make_space(self):
12         now_ts = time.time()
13         delta_ts = now_ts - self.leaking_ts # 距离上一次漏水过去了多久
14         delta_quota = delta_ts * self.leaking_rate # 可以腾出的空间
15         if delta_quota < 1:  # 如果可以腾出的空间太少就算了
16             return
17         self.left_quota += delta_quota  # 增减剩余空间
18         self.leaking_ts = now_ts
19         if self.left_quota > self.self.capacity:  # 剩余空间不能大于漏洞容量
20             self.left_quota = self.capacity
21 
22     def watering(self, quote):
23         self.make_space()
24         if self.left_quota >= quote:
25             self.left_quota -= quote
26             return True
27         return False
28 
29 funnels = {}
30 
31 def is_action_allowed(user_id, action_key, capacity, leaking_rate):
32     key = f'{user_id}:{action_key}'
33     funnel = funnels.get(key)
34     if not funnel:
35         funnel = Funnel(capacity, leaking_rate)
36         funnels[key] = funnel
37     return funnel.watering(1)
38 
39 for i in range(20):
40     print(is_action_allowed('test', 'leak', 15, 0.5))

 

代码逻辑非常清晰，我就不多做解释了。

现在如果我们需要将这个单机版的转变为一个分布式的，首先能够想到的用来存储的结构就是hash了，每次取值都从hash中取出来然后进行运算之后再存回去，但是这三步（取值、运算、存储）我们无法保证其原子性，那么这就需要加锁进行限制，加锁之后带来的问题就是可能加锁失败，然后就得再设计一个等待池进行重试，加锁还会带来性能的下降。。。加锁真的是枷锁。。

那么，有没有更优雅的方法呢？redis大神开发的一个模块redis-cell就解决了这个问题，这个模块使用的也是漏斗限流的方式（我查了以下，也有的叫做令牌桶算法），该模块只有一条指令 cl.throttle ，它的参数和返回值都emmm比较多。

 

下面的示例和上面的代码中的目的是一样的：

>cl.throttle test:leak 15 30 60 1

15就是漏斗容量（capacity），30/60就是速率（leak_rate），1是可选参数，相当于上述代码中37行的的参数1，该参数的默认值也是1

 

其返回值是这样的：

> cl.throttle laoqian:reply 15 30 60 1) (integer) 0 # 0 表示允许，1 表示拒绝 2) (integer) 15 # 漏斗容量 capacity 3) (integer) 14 # 漏斗剩余空间 left_quota 4) (integer) -1 # 如果拒绝了，需要多长时间后再试(漏斗有空间了，单位秒) 5) (integer) 2 # 多长时间后，漏斗完全空出来(left_quota==capacity，单位秒)