大家好，我是dandyhuang。高并发三板斧：限流、缓存、降级。 限流其实就是：当高并发或者瞬时高并发时，为了保证系统的稳定性、可用性，系统以牺牲部分请求为代价或者延迟处理请求为代价，保证系统整体服务可用。今天就给大家介绍golang 官方扩展包 time(golang.org/x/time/rate) 中，一个基于令牌桶的限流器实现。它的实现java guava rate limiter中的实现思路是一样的。按照该思路我也用c++实现了一份代码，可以供大家参考代码token_limiter.cpp实现。
令牌桶
令牌桶算法的原理：系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。自然我们可能会想到启一个协程，定时的往桶里丢入令牌。但golang中没有按照这个思路去实现。而是实时计算当前产生的令牌数。可能会有人觉得，把定时的时间片调低一些，比如1us就去计算产生的令牌数。这种效果也和实时计算效果也差不多。但这种处理，随着精度越高，cpu亲缘性的问题就越严重。我们来看看golang是怎么实现的。思路其实也是比较巧妙。
time/rate实现
package limiter
import (
"fmt"
"testing"
"time"
"golang.org/x/time/rate"
)
func TestLimter(t *testing.T) {
limiter := rate.NewLimiter(rate.Every(time.Millisecond*10), 2)
for i := 0; i < 10; i++ {
var ok bool
if limiter.Allow() {
ok = true
}
time.Sleep(time.Millisecond * 3)
fmt.Println(ok, limiter.Burst())
}
}
执行结果：
=== RUN TestLimter
true 2
true 2
false 2
true 2
false 2
false 2
true 2
false 2
false 2
true 2
--- PASS: TestLimter1 (0.03s)
因为初始化有2个令牌在里头，随着后续执行，每10ms产生一个令牌。所以后续10ms内，只有一个请求能获取到令牌。
创建限流器
func NewLimiter(r Limit, b int) *Limiter {
return &Limiter{
limit: r,
burst: b,
}
}
type Limiter struct {
mu sync.Mutex
limit Limit
burst int
tokens float64
// last is the last time the limiter's tokens field was updated
last time.Time
// lastEvent is the latest time of a rate-limited event (past or future)
lastEvent time.Time
}
Limiter参数中
mu：每次请求并发安全，加锁处理
burst : 桶内能存放令牌的个数
limit：生成令牌的速率
tokens： 剩余令牌个数
last： 上一次取走 token 的时间
lastEvent：上一次限流事件的时间
Allow 判断是否运行通过
// 判断是否满足条件
func (lim *Limiter) Allow() bool {
return lim.AllowN(time.Now(), 1)
}
func (lim *Limiter) AllowN(now time.Time, n int) bool {
return lim.reserveN(now, n, 0).ok
}
func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation {
// 加锁防止并发
lim.mu.Lock()
defer lim.mu.Unlock()
// 边界处理，都是一些异常设置，设置最大值，或没有设置的场景
if lim.limit == Inf {
return Reservation{
ok: true,
lim: lim,
tokens: n,
timeToAct: now,
}
} else if lim.limit == 0 {
var ok bool
if lim.burst >= n {
ok = true
lim.burst -= n
}
return Reservation{
ok: ok,
lim: lim,
tokens: lim.burst,
timeToAct: now,
}
}
// 从上次获取令牌到当前时间，共产生的令牌数tokens
now, last, tokens := lim.advance(now)
// 减去需要的令牌数据
tokens -= float64(n)
var waitDuration time.Duration
// 如果需要的令牌数tokens 为负数，则计算需要等待的 WaitDuration时间
if tokens < 0 {
waitDuration = lim.limit.durationFromTokens(-tokens)
}
// 获取需要的令牌数如果大于桶的容量 或者 等待时间大于设置的maxFutureReserve
ok := n <= lim.burst && waitDuration <= maxFutureReserve
// 构造一个Reservation对象，后续结果返回该对象
r := Reservation{
ok: ok,
lim: lim,
limit: lim.limit,
}
// 成功才更新该对象
if ok {
r.tokens = n
r.timeToAct = now.Add(waitDuration)
}
// 成功了，才把上次时间和tokens数更新
if ok {
lim.last = now
lim.tokens = tokens
lim.lastEvent = r.timeToAct
} else {
// 这里因为锁并发的问题，所以才回导致last会有更新
lim.last = last
}
return r
}
reserveN中记录了上次访问的时间和当前桶中令牌的数量。当再次访问时，通过上次访问记录的时间实时计算当前令牌的数量，决定是否可以放行。
advance计算产生的令牌数
func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
last := lim.last
// 因为加锁的原因，所以出现这种情况
if now.Before(last) {
last = now
}
// （当前时间-上次时间）* 速率 = 产生的令牌数
elapsed := now.Sub(last)
delta := lim.limit.tokensFromDuration(elapsed)
tokens := lim.tokens + delta
if burst := float64(lim.burst); tokens > burst {
tokens = burst
}
return now, last, tokens
}
这里AllowN()请求中，我们是在Allow()的时候调用lim.AllowN(time.Now(), 1)，并把当前时间传入。这时候，如果
reserveN处理的比较慢，并且请求成功。如线程t1请求的时间为10:10:12秒，线程t2请求时间为10:10:10秒。 而线程t1先抢到了锁。并处理请求成功。接下去t2才进行处理。这时last为10:10:12秒，now为10:10:10秒的场景
总结
golang/rate包中，牺牲一点加锁的性能，实时计算产生的令牌数。这种实现的好处： 对令牌的计算可以非常精确。而对比于定时往桶里添加令牌的实现，虽然在请求可以使用原子计算，不上锁实现。但对于令牌的计算来说，是比较不准确的，需要根据定时器的精度来保证。而精度越小，cpu亲缘性问题就越明显。个人觉得虽然加锁的实现，对性能有一部分影响，但是令牌桶都是在计算，所以性能不会有很大的问题，加锁时间不长。
