常见限流算法和go语言time/rate go.uber.org/ratelimit讲解

1 背景

保护服务节点或者数据节点，防止瞬时流量过大造成服务和数据崩溃，导致服务不可用
1

2 主流限流算法

2.1 固定/滑动窗口限流算法

2.1.1 固定窗口限流

实现过程:
也叫计数器算法，顾名思义就是固定一个窗口,在这个时间窗口下维护一个累加的计数器，每来一个请求计数器+1,并判断计数是否超过阈值，
超过阈值则拒绝请求直到下一个时间窗口将计数器清零再继续放行请求

优点: 原理简单 实现也简单
缺点: 限流策略太粗糙，无法限制两个时间窗口临界时的瞬时流量，比如：
        我们设置限流策略为时间窗口为1s 限制请求个数为100，有可能会出现这种情况：
        第1s的100个请求都集中在第500ms-1000ms内，第2s的请求都集中在第0ms-500ms内，那在这2s内，有200个请求发生在了
        第500ms到1500ms这1s内，显然不符合我们原来的预期

1
2
3
4
5
6
7
8
9
10

2.1.2 滑动窗口限流

实现过程:
滑动窗口算法是固定窗口的一种改进，但根本上并没有真正解决固定窗口算法的临界突发流量问题
滑动窗口算法将一个大的时间窗口分成多个小窗口，每次大窗口向后滑动一个小窗口，并保证大的窗口内流量不会超出最大值，这种实现比固定窗口的流量曲线更加平滑

假设还是1s限制100请求，将1s分为10个小窗口，[[t1_start,t1_end], [t2_start, t2_end]...[t10_start, t1_end]]，每个小窗口维护一个自己的计数器
1. 判断请求时间是否在当前大窗口内，是则3
2. 大窗口往后一定一个小窗口 [[t2_start, t2_end]...[t11_start, t11_end]] 执行1
3. 判断大窗口下所有小窗口的计数器之和是否>=100 是则拒绝请求 否则放行，并且小窗口计数器+1

优点: 限流比固定窗口更加平滑
缺点: 虽然解决了临界突发流量问题，但是还是会存在流量突刺，比如 1s限流100 结果100个请求全发生在前10ms内
1
2
3
4
5
6
7
8
9
10
11

2.2 漏桶算法

还有一种漏桶算法，算法内维护一个容器，请求进来时相当于水流进容器，处理请求时相当于水从容器流出。容器有一个最大容量，
并且水匀速从桶内流出。
实现过程:
可以维护一个固定长度队列，新的请求过来后，队列未满则放入队列，队列已满则拒绝请求，
然后再维护一个线程池，定期从队列里获取并处理请求

优点: 没有了流量突刺问题
缺点: 无法应对流量突发问题
1
2
3
4
5
6
7
8

2.3 令牌桶

令牌桶算是漏桶算法的一种改进算法，解决漏桶无法应对突发流量的问题，
想象有一个固定大小的桶，系统会以恒定速率向桶中放 Token，桶满则暂时不放。
而用户则从桶中取Token，如果有剩余token就可以一直取。如果没有剩余token，则需要等到系统中被放置了token才行。

实现过程:
其实我们并不需要真正去维护一个桶或者队列，这样的效率比较低, 比如golang标准库golang.org/x/time/rate，
通过一个计数器即可完成限流。具体实现可以看下面的介绍
1
2
3
4
5
6
7

令牌桶

3 golang标准库库实现限流算法

官方地址: golang.org/x/time/rate

github地址:github.com/golang/time/rate

原文章链接

3.1 构造一个限流器

limiter := NewLimiter(10, 1);
这里有两个参数：
第一个参数是 r Limit。代表每秒可以向token桶中产生多少token。Limit 实际上是 float64 的别名。
第二个参数是 b int。b代表 token 桶的容量大小。
那么，对于以上例子来说，其构造出的限流器含义为，其令牌桶大小为1, 以每秒10个token 的速率向桶中放置 token

除了直接指定每秒产生的 token 个数外，还可以用 Every 方法来指定向 token 桶中放置 token 的间隔，例如：
limit := Every(100 * time.Millisecond);
limiter := NewLimiter(limit, 1);
以上就表示每 100ms 往桶中放一个 token。本质上也就是一秒钟产生 10 个。

Limiter 提供了三类方法供用户消费 token，用户可以每次消费一个 token，也可以一次性消费多个 token。
而每种方法代表了当 token 不足时，各自不同的对应手段
1
2
3
4
5
6
7
8
9
10
11
12
13

3.2 Wait/WaitN

func (lim *Limiter) Wait(ctx context.Context) (err error)
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)
Wait 实际上就是 WaitN(ctx,1)。

当使用 Wait 方法消费 token 时，如果此时桶内 token 数组不足 (小于 N)，那么 Wait 方法将会阻塞一段时间，直至 token 满足条件。如果充足则直接返回。

这里可以看到，Wait 方法有一个 context 参数。
我们可以设置 context 的 Deadline 或者 Timeout，来决定此次 Wait 的最长时间
1
2
3
4
5
6
7
8

3.3 Allow/AllowN

func (lim *Limiter) Allow() bool
func (lim *Limiter) AllowN(now time.Time, n int) bool
Allow 实际上就是 AllowN(time.Now(),1)。

AllowN 方法表示，截止到某一时刻，目前桶中数目是否至少为 n 个，满足则返回 true，同时从桶中消费 n 个 token。
反之返回不消费 token，false。

通常对应这样的线上场景，如果请求速率过快，就直接丢到某些请求。
1
2
3
4
5
6
7
8

3.4 Reserve/ReserveN

func (lim *Limiter) Reserve() *Reservation
func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation
Reserve 相当于 ReserveN(time.Now(), 1)。

ReserveN 的用法就相对来说复杂一些，当调用完成后，无论 token 是否充足，都会返回一个 Reservation * 对象。

你可以调用该对象的 Delay() 方法，该方法返回了需要等待的时间。如果等待时间为 0，则说明不用等待。
必须等到等待时间之后，才能进行接下来的工作。

或者，如果不想等待，可以调用 Cancel() 方法，该方法会将 token 归还。
1
2
3
4
5
6
7
8
9
10

3.5 动态调整速率

Limiter 支持可以调整速率和桶大小：

SetLimit(Limit) 改变放入 token 的速率
SetBurst(int) 改变 token 桶大小
有了这两个方法，可以根据现有环境和条件，根据我们的需求，动态的改变 token 桶大小和速率
1
2
3
4
5

3.6 Token 的生成和消费

在 time/rate 中，NewLimiter 的第一个参数是速率 limit，代表了一秒钟可以产生多少 token。
那么简单换算一下，我们就可以知道一个token的生成间隔是多少。

有了这个生成间隔，我们就可以轻易地得到两个数据：
1. 生成 N 个新的token一共需要多久。time/rate 中对应的实现函数为 durationFromTokens。
2. 给定一段时长，这段时间一共可以生成多少个 token。time/rate 中对应的实现函数为 tokensFromDuration

那么，有了这些转换函数，整个过程就很清晰了，如下：

计算从上次取token的时间到当前时刻, 期间一共新产生了多少 token：
我们只在取token之前生成新的token，也就意味着每次取token的间隔，实际上也是生成token的间隔。
我们可以利用tokensFromDuration, 轻易的算出这段时间一共产生 token 的数目。
那么，当前token数目 = 新产生的token数目 + 之前剩余的token数目 - 要消费的token数目。

如果消费后剩余token数目大于零，说明此时token桶内仍不为空，此时token充足，无需调用侧等待。
如果token数目小于零，则需等待一段时间。
那么这个时候，我们可以利用durationFromTokens将当前负值的token数转化为需要等待的时间。将需要等待的时间等相关结果返回给调用方。

此外，整个过程为了保证线程安全，更新令牌桶相关数据时都用了 mutex 加锁

我们模拟下请求与token数变化的关系：

当某一时间，桶内token数为3, 此时A线程请求5个token。那么此时桶内token不足，因此A线程需要等待2个token的时间。且此时桶内token数变为-2。
同时，B线程请求4个token，此时桶内token数为-2，因此B线程需要等待2+4=6个token的时间，且此时桶内token数变为-6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

3.7 float 精度问题

从上面原理讲述可以看出，在 token 和时间的相互转化函数 durationFromTokens 和 tokensFromDuration 中，涉及到 float64 的乘除运算。
一谈到 float 的乘除，我们就需要小心精度问题了。

而 Golang 在这里也踩了坑，以下是 tokensFromDuration 最初的实现版本

func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	return d.Seconds() * float64(limit)
}
这个操作看起来一点问题都没：每秒生成的 token 数乘于秒数。
然而，这里的问题在于，d.Seconds() 已经是小数了。两个小数相乘，会带来精度的损失。

所以就有了这个 issue:golang.org/issues/34861。

修改后新的版本如下：

func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	sec := float64(d/time.Second) * float64(limit)
	nsec := float64(d%time.Second) * float64(limit)
	return sec + nsec/1e9
}
time.Duration 是 int64 的别名，代表纳秒。分别求出秒的整数部分和小数部分，进行相乘后再相加，这样可以得到最精确的精度
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

3.8 数值溢出问题

我们讲 reserveN 函数的具体实现时，第一步就是计算从当前时间到上次取 token 的时刻，期间一共新产生了多少 token，同时也可得出当前的 token 是多少。

我最开始的理解是，直接可以这么做：

// elapsed 表示过去的时间差
elapsed := now.Sub(lim.last)
// delta 表示这段时间一共新产生了多少 token
delta = tokensFromDuration(now.Sub(lim.last))

tokens := lim.tokens + delta
if(token> lim.burst){
	token = lim.burst
}
其中，lim.tokens 是当前剩余的 token，lim.last 是上次取 token 的时刻。lim.burst 是 token 桶的大小。
使用 tokensFromDuration 计算出新生成了多少 token，累加起来后，不能超过桶的容量即可。

这么做看起来也没什么问题，然而并不是这样。

在 time/rate 里面是这么做的，如下代码所示：

maxElapsed := lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)
elapsed := now.Sub(last)
if elapsed > maxElapsed {
	elapsed = maxElapsed
}

delta := lim.limit.tokensFromDuration(elapsed)

tokens := lim.tokens + delta
if burst := float64(lim.burst); tokens > burst {
	tokens = burst
}
与我们最开始的代码不一样的是，它没有直接用 now.Sub(lim.last) 来转化为对应的 token 数，而是
先用 lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)，计算把桶填满的时间 maxElapsed。
取 elapsed 和 maxElapsed 的最小值。

这么做算出的结果肯定是正确的，但是这么做相比于我们的做法，好处在哪里？

对于我们的代码，当 last 非常小的时候（或者当其为初始值 0 的时候），此时 now.Sub(lim.last) 的值就会非常大，如果 lim.limit 即每秒生成的 token 数目也非常大时，直接将二者进行乘法运算，** 结果有可能会溢出。**

因此，time/rate 先计算了把桶填满的时间，将其作为时间差值的上限，这样就规避了溢出的问题
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

3.9 token的归还

而对于 Reserve 函数，返回的结果中，我们可以通过 Reservation.Delay() 函数，得到需要等待时间。
同时调用方可以根据返回条件和现有情况，可以调用 Reservation.Cancel() 函数，取消此次消费。
当调用 Cancel() 函数时，消费的 token 数将会尽可能归还给 token 桶。

此外，我们在 上一篇文章 中讲到，Wait 函数可以通过 Context 进行取消或者超时等，
当通过 Context 进行取消或超时时，此时消费的 token 数也会归还给 token 桶。

然而，归还 token 的时候，并不是简单的将 token 数直接累加到现有 token 桶的数目上，这里还有一些注意点：

restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))
if restoreTokens <= 0 {
	return
}
以上代码就是计算需要归还多少的 token。其中：

r.tokens 指的是本次消费的 token 数
r.timeToAct 指的是 token 桶可以满足本次消费数目的时刻，也就是消费的时刻 + 等待的时长。
r.lim.lastEvent 指的是最近一次消费的 timeToAct 值
其中：r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct)) 指的是，从该次消费到当前时间，一共又新消费了多少 token 数目。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

4 uber-go官方库限流

官方地址: go.uber.org/ratelimit

github地址: github.com/uber-go/ratelimit

原文章链接

4.1 ratelimit 的使用

我们直接看下 uber-go 官方库给的例子：

rl := ratelimit.New(100) // per second

prev := time.Now()
for i := 0; i < 10; i++ {
  now := rl.Take()
  fmt.Println(i, now.Sub(prev))
  prev = now
}
在这个例子中，我们给定限流器每秒可以通过 100 个请求，也就是平均每个请求间隔 10ms。
因此，最终会每 10ms 打印一行数据。输出结果如下：

// Output:
// 0 0
// 1 10ms
// 2 10ms
// 3 10ms
// 4 10ms
// 5 10ms
// 6 10ms
// 7 10ms
// 8 10ms
// 9 10ms

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

4.2 基本实现

要实现以上每秒固定速率的目的，其实还是比较简单的。

在 ratelimit 的 New 函数中，传入的参数是每秒允许请求量 (RPS)。
我们可以很轻易的换算出每个请求之间的间隔：

limiter.perRequest = time.Second / time.Duration(rate)
以上 limiter.perRequest 指的就是每个请求之间的间隔时间。

如下图，当请求 1 处理结束后, 我们记录下请求 1 的处理完成的时刻, 记为 limiter.last。
稍后请求 2 到来, 如果此刻的时间与 limiter.last 相比并没有达到 perRequest 的间隔大小，那么 sleep 一段时间即可。
1
2
3
4
5
6
7
8
9
10

对应 ratelimit 的实现代码如下：

sleepFor = t.perRequest - now.Sub(t.last)
if sleepFor > 0 {
	t.clock.Sleep(sleepFor)
	t.last = now.Add(sleepFor)
} else {
	t.last = now
}
1
2
3
4
5
6
7
8
9

4.3 最大松弛量

我们讲到，传统的 Leaky Bucket，每个请求的间隔是固定的，然而，在实际上的互联网应用中，流量经常是突发性的。对于这种情况，uber-go 对 Leaky Bucket 做了一些改良，引入了最大松弛量 (maxSlack) 的概念。

我们先理解下整体背景: 假如我们要求每秒限定 100 个请求，平均每个请求间隔 10ms。但是实际情况下，有些请求间隔比较长，有些请求间隔比较短。如下图所示：
1
2
3

请求 1 完成后，15ms 后，请求 2 才到来，可以对请求 2 立即处理。请求 2 完成后，5ms 后，请求 3 到来，这个时候距离上次请求还不足 10ms，因此还需要等待 5ms。

但是，对于这种情况，实际上三个请求一共消耗了 25ms 才完成，并不是预期的 20ms。在 uber-go 实现的 ratelimit 中，可以把之前间隔比较长的请求的时间，匀给后面的使用，保证每秒请求数 (RPS) 即可。

对于以上 case，因为请求 2 相当于多等了 5ms，我们可以把这 5ms 移给请求 3 使用。加上请求 3 本身就是 5ms 之后过来的，一共刚好 10ms，所以请求 3 无需等待，直接可以处理。此时三个请求也恰好一共是 20ms。
如下图所示：
1
2
3
4
5
6
7

在 ratelimit 的对应实现中很简单，是把每个请求多余出来的等待时间累加起来，以给后面的抵消使用。

t.sleepFor += t.perRequest - now.Sub(t.last)
if t.sleepFor > 0 {
  t.clock.Sleep(t.sleepFor)
  t.last = now.Add(t.sleepFor)
  t.sleepFor = 0
} else {
  t.last = now
}
注意：这里跟上述代码不同的是，这里是 +=。而同时 t.perRequest - now.Sub(t.last) 是可能为负值的，负值代表请求间隔时间比预期的长。

当 t.sleepFor > 0，代表此前的请求多余出来的时间，无法完全抵消此次的所需量，因此需要 sleep 相应时间, 同时将 t.sleepFor 置为 0。

当 t.sleepFor < 0，说明此次请求间隔大于预期间隔，将多出来的时间累加到 t.sleepFor 即可。

但是，对于某种情况，请求 1 完成后，请求 2 过了很久到达 (好几个小时都有可能)，那么此时对于请求 2 的请求间隔 now.Sub(t.last)，会非常大。以至于即使后面大量请求瞬时到达，也无法抵消完这个时间。那这样就失去了限流的意义。

为了防止这种情况，ratelimit 就引入了最大松弛量 (maxSlack) 的概念, 该值为负值，表示允许抵消的最长时间，防止以上情况的出现。

if t.sleepFor < t.maxSlack {
  t.sleepFor = t.maxSlack
}
ratelimit 中 maxSlack 的值为 -10 * time.Second / time.Duration(rate), 是十个请求的间隔大小。我们也可以理解为 ratelimit 允许的最大瞬时请求为 10。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

4.4 高级用法

ratelimit 的 New 函数，除了可以配置每秒请求数 (QPS)， 其实还提供了一套可选配置项 Option。

func New(rate int, opts ...Option) Limiter
Option 的类型为 type Option func(l *limiter), 也就是说我们可以提供一些这样类型的函数，作为 Option，传给 ratelimit, 定制相关需求。

但实际上，自定义 Option 的用处比较小，因为 limiter 结构体本身就是个私有类型，我们并不能拿它做任何事情。

我们只需要了解 ratelimit 目前提供的两个配置项即可：
1
2
3
4
5
6
7
8

• WithoutSlack

我们上文讲到 ratelimit 中引入了最大松弛量的概念，而且默认的最大松弛量为 10 个请求的间隔时间。

但是确实会有这样需求场景，需要严格的限制请求的固定间隔。那么我们就可以利用 WithoutSlack 来取消松弛量的影响。

limiter := ratelimit.New(100, ratelimit.WithoutSlack)
1
2
3
4
5

• WithClock(clock Clock)

我们上文讲到，ratelimit 的实现时，会计算当前时间与上次请求时间的差值，并 sleep 相应时间。
在 ratelimit 基于 go 标准库的 time 实现时间相关计算。如果有精度更高或者特殊需求的计时场景，可以用 WithClock 来替换默认时钟。

通过该方法，只要实现了 Clock 的 interface，就可以自定义时钟了。

type Clock interface {
	Now() time.Time
	Sleep(time.Duration)
}
clock &= MyClock{}
limiter := ratelimit.New(100, ratelimit.WithClock(clock))
1
2
3
4
5
6
7
8
9
10
11