本文主要介绍两个包Uber漏桶,time/rate令牌桶 可以了解到:
- 使用方法
- 漏桶/令牌桶 两种限流思想 and 实现原理
- 区别及适用场景
- 应用Case
背景
我们为了保护系统资源,防止过载,常常会使用限流器。
使用场景:
- API速率限制:防止用户恶意刷接口(当然这部分现在有公司风控部门来管控,例如whale);
- 刷数据:控制对数据库的访问速率,避免瞬时大流量打崩数据库,保护实例;
- 服务降级:例如节假日对一些非关键服务进行服务降级,保证关键服务的可用性;
- 爬虫:大多数网站具有反爬能力,需要控制爬虫速率;
限流器通用实现方案
- 固定窗口限流:
在这种方法中,时间被分割成固定的窗口,每个窗口都有一个请求的最大限制。这种方法简单易实现,但可能在窗口交界处出现请求的瞬时峰值。 - 滑动窗口限流:
滑动窗口限流是对固定窗口限流的改进,它将时间分割成更小的窗口,从而使请求分布更加均匀。这种方法可以更好地控制请求的速率,但实现起来更复杂。 - 并发限流:
并发限流是限制系统中同时处理的请求的数量。这种方法可以防止系统过载,但需要注意防止长时间的请求阻塞其他请求。 - 漏桶限流(Leaky Bucket) :
系统将请求放入桶中,然后以固定的速率从桶中取出请求进行处理,就像水从漏桶中以固定的速率漏出一样。如果桶已满,则新的请求会等待。漏桶的优点是输出数据的速率总是恒定的,无论输入数据的速率如何变化。 - 令牌桶限流(Token Bucket):
系统以固定的速率向桶中添加令牌,请求在处理前需要从桶中获取令牌,只有获取到令牌的请求才会被处理。如果桶中没有足够的令牌,则请求需要等待或被拒绝。令牌桶的优点是能够应对突发流量,因为在低流量时期,未使用的令牌可以积攒起来,用于应对突发的高流量。 - 自适应限流
根据系统能够承载的最大吞吐量,来进行限流,当当前的流量大于最大吞吐的时候就限制流量进入,反之则允许通过;需要一个机器指标来做触发阈值,常见的选择一般是机器负载、CPU 使用率,总体平均响应时间、入口 QPS 和并发线程数等。
本文我们主要聚焦于 漏桶限流 和 令牌桶限流。这两种方法分别有两种代表性的包
- 漏桶限流:Uber开源的第三方包,目前4k Stars,MIT license 主要用于服务限流GitHub - uber-go/ratelimit: A Go blocking leaky-bucket rate limit implementation
- 令牌桶限流:Go标准库,应用广泛golang.org/x/time/rate
Uber “改良版”的漏桶限流
漏桶的基本思想是将该组件想象成一个出口大小固定的桶,输出的数据流只能以固定的速率输出。
GitHub - uber-go/ratelimit: A Go blocking leaky-bucket rate limit implementation 包采用的是“改良版的漏桶”方式实现。
为什么这么说呢?因为有两部分做了相关的改进:
漏桶实现方案 | “改良版的漏桶”方式实现 |
---|---|
待处理队列Queue | 无待处理队列,直接处理 or 等待(或者可以理解为待处理队列长度=1) |
最大输出速率固定 | 通过松弛的方式保证整体上的QPS,输出速率可能瞬时增高,允许短时间的突发流量 |
1. 使用方法
1.1 初始化
rl := ratelimit.New(100) // per second
第一个参数代表每秒可以处理多少个请求。
1.2 限流
now := rl.Take()
如果不满足条件(请求速率过快),Take方法将会阻塞一段时间,直至满足条件。如果时间条件满足则直接返回。(可以类比下文标准库中的 Wait 方法)
1.3 举个🌰
import ("fmt""time""go.uber.org/ratelimit"
)func main() {rl := ratelimit.New(100) // per secondprev := time.Now()for i := 0; i < 10; i++ {now := rl.Take()fmt.Println(i, now.Sub(prev))prev = now}
}
// Output
0 0s
1 10ms
2 10ms
3 10ms
4 10ms
5 10ms
6 10ms
7 10ms
8 10ms
9 10ms
可以看到,每次执行的间隔时间为10ms
2. 原理分析
(下文介绍2023.7更新的V3版本,本人认为该思路更好理解,网上也有许多介绍V1/V2版本的做法,参考文档)
2.1 初始化New方法
// New returns a Limiter that will limit to the given RPS.
func New(rate int, opts ...Option) Limiter {return newAtomicInt64Based(rate, opts...)
}type Limiter interface {// Take should block to make sure that the RPS is met.Take() time.Time
}
创建了一个Limiter Interface,库中有四个实现,如下所示:
- atomicInt64Limiter: 使用sync/atomic并发安全+高性能的限流版本,V3版本限流
- atomicLimiter: 使用sync/atomic并发安全的限流版本
- mutexLimiter: 使用sync.Mutex并发安全的限流版本
- unlimited: 不限流
2.2 atomicInt64Limiter 对象
type atomicInt64Limiter struct {prepadding [64]byte // 防止伪共享缓存state int64 // 理论上本次应该执行的时间,从0开始累计(纳秒)postpadding [56]byte // 防止伪共享缓存perRequest time.Duration // 每次请求理论间隔时间maxSlack time.Duration // 最大松弛量,默认值是10个请求理论间隔时间(V3版本这里是正值)clock Clock // 休眠时间
}// 初始化
func newAtomicInt64Based(rate int, opts ...Option) *atomicInt64Limiter {config := buildConfig(opts)perRequest := config.per / time.Duration(rate)l := &atomicInt64Limiter{perRequest: perRequest,maxSlack: time.Duration(config.slack) * perRequest,clock: config.clock,}// 原子操作,保证对state值修改过程的并发安全atomic.StoreInt64(&l.state, 0)return l
}// 默认一些通用的配置
func buildConfig(opts []Option) config {c := config{clock: clock.New(),slack: 10,per: time.Second,}for _, opt := range opts {opt.apply(&c)}return c
}
atomicInt64Limiter 对象中,prepadding和postpadding 字段,主要是用于填充 CPU 缓存行,防止伪共享缓存的。不了解的朋友可以参考这篇文章 CPU缓存体系对Go程序的影响。
2.3 限流Take 方法
该方法可以阻塞请求,保证每次请求的时间间隔。
func (t *atomicInt64Limiter) Take() time.Time {var (newTimeOfNextPermissionIssue int64 // 下一次请求理论执行时间now int64 // 当前时间,纳秒)for {now = t.clock.Now().UnixNano()timeOfNextPermissionIssue := atomic.LoadInt64(&t.state) // 上一次理论执行时间switch {case timeOfNextPermissionIssue == 0 || (t.maxSlack == 0 && now-timeOfNextPermissionIssue > int64(t.perRequest)):// 第一个请求 or 无最大松弛时间,不控制速度newTimeOfNextPermissionIssue = nowcase t.maxSlack > 0 && now-timeOfNextPermissionIssue > int64(t.maxSlack):// 当前请求-上一次请求时间中间的间隔过长,大于最大松弛时间,下一次理论执行时间定义为now-最大松弛时间,即保留了最大松弛时间的长度用来调整newTimeOfNextPermissionIssue = now - int64(t.maxSlack)default:// 最正常的情况,下一次请求理论执行时间 = 上一次理论执行时间 + 理论实践间隔newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest)}// 更新t.state 执行时间的状态,改为newTimeOfNextPermissionIssue,如果修改成功,跳出死循环,保证并发安全if atomic.CompareAndSwapInt64(&t.state, timeOfNextPermissionIssue, newTimeOfNextPermissionIssue) {break}}// 计算需要休眠的时间,如果是正数,那么需要休眠对应的时间sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now)if sleepDuration > 0 {t.clock.Sleep(sleepDuration)return time.Unix(0, newTimeOfNextPermissionIssue)}return time.Unix(0, now)
}
刚看到这么复杂的逻辑,一定心里发怵,但是说实话已经比V2版本的清晰多了。下面画个图跟大家讲一下。
3. Case分析
假定限定 100ms 执行一个请求
-
Case1:如果请求2提前到来
-
结论:会等待到理论时间间隔再执行
1. 请求1到来,命中switch中第一个条件 case timeOfNextPermissionIssue == 0 ,为(t *atomicInt64Limiter).State = now = 0,sleepDuration=0,请求1直接执行
2. 请求2在50ms时到来,命中switch中default条件,计算得到请求2的理论执行时间newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest) = 0+100ms = 100ms并更新(t *atomicInt64Limiter).State =newTimeOfNextPermissionIssus = 100ms,由于sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now) = 100ms - 50ms = 50ms,系统休眠50ms,即阻塞执行,最后返回请求2实际执行时间 100ms -
Case2:如果请求2延迟到来,请求3提前到来
-
结论:请求2直接执行,请求3仅需要等待到200ms时执行,使三个请求整体上看间隔时间平均100ms,具有一定的松弛度
1. 请求1到来,同上,(t *atomicInt64Limiter).State = now = 0,sleepDuration=0,请求1直接执行
2. 请求2在130ms时到来,命中switch中default条件,计算得到请求2的理论执行时间newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest) = 0+100ms = 100ms并更新(t *atomicInt64Limiter).State =newTimeOfNextPermissionIssus = 100ms,由于sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now) = 100ms - 130ms = -30ms,休眠时间为负数,即请求2直接在到来时刻130ms时执行
3. 请求3在180ms时到来,命中switch中default条件,计算得到请求2的理论执行时间newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest) = 100ms+100ms = 200ms并更新(t *atomicInt64Limiter).State =newTimeOfNextPermissionIssus = 200ms,由于sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now) = 200ms - 180ms = 20ms,休眠时间为正数,系统休眠20ms,即阻塞执行,最后返回请求3实际执行时间 200ms,这里比较关键,请求3与请求2的间隔实际为200ms-130ms=70ms,并不是理论上每个间隔严格保证100ms,仅是保证整体上间隔平均时间为100ms -
Case3:如果请求2与请求1间隔时间非常长,请求3及其后续密集到来
这个场景非常极端,由于请求2-请求1过长,后续都会在请求发送时刻执行,导致后续会限流失效,因为并没有违背整体上的10QPS的要求。
所以这里引入了一个概念最大松弛时间(10倍的请求间隔时间),用来保证后续限流的正常执行。 -
结论:请求2直接执行,请求3-请求12在请求发送时刻执行,请求13休眠,开始限流。
1. 请求2在2100ms时到来,命中switch中t.maxSlack > 0 && now-timeOfNextPermissionIssue > int64(t.maxSlack)(2100ms-0>1000ms)条件,给予最大松弛时间的额度,newTimeOfNextPermissionIssue = now - int64(t.maxSlack) = 2100ms - 1000ms = 1100ms,远大于正常用default计算出来的100ms。由于sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now) = 1100ms - 2100ms = -1000ms,休眠时间为负数,即请求2直接在到来时刻2100ms时执行
2. 请求3在2105ms到来,命中switch中的default条件,newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest)=1100ms+100ms = 1200ms,由于sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now) = 1200ms - 2105ms = -905ms,休眠时间为负数,即请求3直接在到来时刻2105ms时执行
3. 请求4-请求12与请求3类似,命中switch中的default条件,直接执行。(假设间隔5ms到来)
4. 请求13在2155ms到来,命中switch中的default条件,newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest)=1200ms+10*100ms = 2200ms,由于sleepDuration := time.Duration(newTimeOfNextPermissionIssue - now) = 2200ms - 2155ms = 45ms,休眠45ms,在2200ms时执行
time/rate令牌桶Plus
标准令牌桶方案,想象有一个固定大小的桶,系统会以恒定速率向桶中放Token,桶满则暂时不放。 而用户则从桶中取Token,如果有剩余Token就可以一直取。如果没有剩余Token,则需要等到系统中被放置了Token才行。
golang.org/x/time/rate 包也做了对应的改进
令牌桶实现方案 | “改良版的令牌桶”方式实现 |
---|---|
待处理队列Queue | 无待处理队列,直接处理 or 等待(或者可以理解为待处理队列长度=1) |
定时产生token | 懒加载方式,新请求到来时才根据时间差更新token数目 |
1. 使用方法
1.1 初始化
limiter := NewLimiter(10, 1);
这里有两个参数:
- 第一个参数是r Limit,(QPS)代表每秒可以向Token桶中产生多少token。
- 第二个参数是b int。b代表Token桶的容量大小。(通过设置较大的 桶容量 值,服务可以在短时间内处理大量的请求,从而更好地应对请求峰值。然后,当请求峰值过去后,你的服务可以以 limit 指定的速率继续处理请求,从而避免长期占用过多的系统资源——换句话说这个参数就是“瞬时能处理的上限”)
1.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)。
- 该方法可以类比上文中的Take方法。
- 当使用Wait方法消费Token时,如果此时桶内Token数组不足(小于N),那么Wait方法将会阻塞一段时间,直至Token满足条件。如果充足则直接返回。
- 这里可以看到,Wait方法有一个context参数。 我们可以设置context的Deadline或者Timeout,来决定此次Wait的最长时间。
- Allow/AllowN
func (lim *Limiter) Allow() bool
func (lim *Limiter) AllowN(now time.Time, n int) bool
- 跟 wait 方法的区别是,返回了bool值,如果为true,则允许本次操作,如果为false,则不允许本次操作
- Reserve/ReserveN
func (lim *Limiter) Reserve() *Reservation
func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation
- 必定会返回结构体 *Reservation,通过它的 Delay() 方法判断等待时间
1.3 举个🌰
控制循环的速率,保证每秒只能处理1个,高峰期最高能同时处理3个。
针对Allow方法,可以看到最开始同时处理了三个请求,后两个因为没有令牌导致不处理;
针对Wait和Reserve方法,可以看到每次间隔1s左右即可执行一个。
package mainimport ("fmt""golang.org/x/time/rate""time"
)func main() {limiter := rate.NewLimiter(1, 3) // 每秒生成1个令牌,桶的容量为3// 使用 Allow 方法for i := 0; i < 5; i++ {if limiter.Allow() {fmt.Println("Allow: Allowed")} else {fmt.Println("Allow: Not allowed")}}// 使用 Wait 方法for i := 0; i < 5; i++ {lastTime := time.Now()if err := limiter.Wait(context.Background()); err != nil {fmt.Println("Error:", err)} else {fmt.Printf("Wait: Need to wait for %v\n", time.Now().Sub(lastTime))lastTime = time.Now()fmt.Println("Wait: Allowed")}}// 使用 Reserve 方法for i := 0; i < 5; i++ {reservation := limiter.Reserve()if !reservation.OK() {fmt.Println("Reserve: 桶数量需要大于0")} else {delay := reservation.Delay()fmt.Printf("Reserve: Need to wait for %v\n", delay)time.Sleep(delay)fmt.Println("Reserve: Allowed")}}
}
Output:
Allow: Allowed
Allow: Allowed
Allow: Allowed
Allow: Not allowed
Allow: Not allowed
Wait: wait 1.000398173s
Wait: Allowed
Wait: wait 999.442232ms
Wait: Allowed
Wait: wait 1.000598372s
Wait: Allowed
Wait: wait 999.837132ms
Wait: Allowed
Wait: wait 999.69229ms
Wait: Allowed
Reserve: wait 999.572574ms
Reserve: Allowed
Reserve: wait 999.376376ms
Reserve: Allowed
Reserve: wait 999.708594ms
Reserve: Allowed
Reserve: wait 999.769212ms
Reserve: Allowed
Reserve: wait 999.736995ms
Reserve: Allowed
2. 原理分析
2.1 初始化NewLimiter方法
func NewLimiter(r Limit, b int) *Limiter {return &Limiter{limit: r,burst: b,}
}type Limiter struct {mu sync.Mutex // 锁,并发安全limit Limit // 每秒产生多少token,(QPS)burst int // 桶的最大值tokens float64 // 此刻桶内token数量// last is the last time the limiter's tokens field was updatedlast time.Time // 上一次tokens更新的时间// lastEvent is the latest time of a rate-limited event (past or future)lastEvent time.Time
}
2.2 限流方法
包中提供的限流算法本质都是一个reserveN
为了方便理解,我们重点看Wait方法(更方便和上文Take方法作比较)
func (lim *Limiter) Wait(ctx context.Context) (err error) {return lim.WaitN(ctx, 1)
}func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {// The test code calls lim.wait with a fake timer generator.// This is the real timer generator.newTimer := func(d time.Duration) (<-chan time.Time, func() bool, func()) {timer := time.NewTimer(d)return timer.C, timer.Stop, func() {}}return lim.wait(ctx, n, time.Now(), newTimer)
}// Params:
// ctx 用于计算最长超时不超过ctx存活时间
// n 每一个请求需要多少令牌数,一般为1
// t 当前时间
// newTimer 休眠定时函数
func (lim *Limiter) wait(ctx context.Context, n int, t time.Time, newTimer func(d time.Duration) (<-chan time.Time, func() bool, func())) error {// 赋值上锁lim.mu.Lock()burst := lim.burstlimit := lim.limitlim.mu.Unlock()// 边界校验if n > burst && limit != Inf {return fmt.Errorf("rate: Wait(n=%d) exceeds limiter's burst %d", n, burst)}// 根据ctx判断需要等待的最长时间select {case <-ctx.Done():return ctx.Err()default:}waitLimit := InfDurationif deadline, ok := ctx.Deadline(); ok {waitLimit = deadline.Sub(t)}// 调用的核心方法,返回Reservation对象r := lim.reserveN(t, n, waitLimit)if !r.ok {return fmt.Errorf("rate: Wait(n=%d) would exceed context deadline", n)}// 如果需要限流,休眠对应的时间delay := r.DelayFrom(t) // 下次理论执行时间 - 当前时间if delay == 0 {return nil}ch, stop, advance := newTimer(delay)defer stop()advance() // only has an effect when testingselect {case <-ch:return nilcase <-ctx.Done():r.Cancel()return ctx.Err()}
}
进一步观察reserveN方法,它最后会返回Reservation对象
type Reservation struct {ok bool // 是否预占令牌成功,可以理解成是否执行成功lim *Limiter // 额外存储一份外面的限流结构体tokens int // 本次请求需要的token数量timeToAct time.Time // 请求执行的时间// This is the Limit at reservation time, it can change later.limit Limit // QPS,每秒往桶中放置的数量,Limit是float64的别名
}// Params
// t 当前时间
// n 每一个请求需要多少令牌数,一般为1
// maxFutureReserve 最长等待时间
func (lim *Limiter) reserveN(t 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: t,}} else if lim.limit == 0 {var ok boolif lim.burst >= n {ok = truelim.burst -= n}return Reservation{ok: ok,lim: lim,tokens: lim.burst,timeToAct: t,}}// 计算得到从当前时间桶内tokens数,具体见下面t, tokens := lim.advance(t)// 减去本次请求需要的token数,当前桶内还剩余的tokens数tokens -= float64(n)// 如果令牌数为负数,说明需要进行等待。// 根据令牌数(token/limit)计算等待时长var waitDuration time.Durationif tokens < 0 {waitDuration = lim.limit.durationFromTokens(-tokens)}// 边界校验,是否预占成功ok := n <= lim.burst && waitDuration <= maxFutureReserver := Reservation{ok: ok,lim: lim,limit: lim.limit,}if ok {r.tokens = nr.timeToAct = t.Add(waitDuration) // 需要等待的时间// Update statelim.last = tlim.tokens = tokenslim.lastEvent = r.timeToAct}return r
}
advance计算得到 此刻桶内tokens数
func (lim *Limiter) advance(t time.Time) (newT time.Time, newTokens float64) {last := lim.lastif t.Before(last) {last = t}// ((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = 当前桶内tokens数elapsed := t.Sub(last)delta := lim.limit.tokensFromDuration(elapsed)tokens := lim.tokens + deltaif burst := float64(lim.burst); tokens > burst { // 不超过最大桶内tokens数量tokens = burst}return t, tokens
}
tokensFromDuration计算一段时间内产生的tokens数量
// 核心计算 tokens 数量的函数,秒数*每秒产生的tokens数
func (limit Limit) tokensFromDuration(d time.Duration) float64 {if limit <= 0 {return 0}return d.Seconds() * float64(limit)
}
总结一下:
- 懒加载:每次请求时,根据时间差(本次减去上次保存的时间戳位置)计算 Token 数目,判断是否需要等待
- 使用sync.Mutex锁保证并发安全
- 当令牌桶数量>请求所需的令牌数时,可以获取令牌。
- 当令牌桶数量<请求所需的令牌数时,需要看等待时间先到还是ctx超时时间哪个短。
- 如果等待时间短,可以获取令牌。
- 如果ctx超时时间短,则无法获取令牌,需要回退令牌。
3. Case分析
分析Wait方法,假定限定 100ms 执行一个请求(QPS=10),桶容量为1
-
Case1:如果请求2提前到来
-
结论:会等待到理论时间间隔再执行
1. 请求1到来,lim.advance(t)返回此刻tokens数为1,消耗一个token,不需要等待。
2. 请求2在50ms时到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (50ms-0)/ 1000 * 10+0 = 0.5 ,即当前时刻tokens=0.5,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = -0.5,waitDuration = lim.limit.durationFromTokens(-tokens)=0.5 / 10 s= 50ms,通过r.DelayFrom(t)方法得到需要等待50ms,最后返回请求2实际执行时间 100ms -
Case2:如果请求2延迟到来,请求3提前到来
-
结论:请求2直接执行,请求3仍需要等待(与请求2间隔100ms)再执行——由于没有松弛度的概念,跟Uber方法的区别
1. 请求1直接执行
2. 请求2在50ms时到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (130ms-0)/ 1000 * 10+0 = 1.3 ,即当前时刻tokens=1.3,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = 0.3,不需要等待,直接执行。
3. 请求3在180ms时到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (180ms-130ms)/ 1000 * 10+0 = 0.5 ,即当前时刻tokens=0.5,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = -0.5,waitDuration = lim.limit.durationFromTokens(-tokens)=0.5 / 10 s= 50ms,通过r.DelayFrom(t)方法得到需要等待50ms,最后返回请求3实际执行时间 230ms -
Case3:桶容量=3,短时间内来了5个请求(每隔5ms)
-
结论:请求1-3直接执行,请求4-5需要等待——令牌桶的优越性
1. 请求1在0ms到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (0-0)/ 1000 * 10+3 = 3 ,即当前时刻tokens=3,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = 2,不需要等待,直接执行。
2. 请求2在5ms时到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (5ms-0)/ 1000 * 10+2 = 2.05 ,即当前时刻tokens=2.05,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = 1.05,不需要等待,直接执行。
3. 请求3在10ms时到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (10ms-5ms)/ 1000 * 10+1.05 = 1.1 ,即当前时刻tokens=1.1,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = 0.1,不需要等待,直接执行。
4. 请求4在15ms时到来,lim.advance(t)计算逻辑,((当前时间 - 上次请求时间) * 每秒产生的tokens数量)+ 桶内原有tokens数 = (15ms-10ms)/ 1000 * 10+0.1 = 0.15 ,即当前时刻tokens=0.15,tokens -= float64(n) 消耗掉本次token需要的数目,tokens = -0.85,waitDuration = lim.limit.durationFromTokens(-tokens)=0.85 / 10 *1000ms= 85ms,通过r.DelayFrom(t)方法得到需要等待85ms,最后返回请求3实际执行时间 100ms
对比一下
共同点
- 无待处理队列,直接处理 or 等待(或者可以理解为待处理队列长度=1)
- 可以支持限制请求的固定间隔
- 并发安全
不同点
Uber | time/rate | |
---|---|---|
限流结构体 | 较为简单 (设计了防止伪共享缓存的字段) | 较为复杂 |
计算方式 | 时间转化为int64,仅涉及加减 | float64形式计算,涉及乘法,成本较高 |
计算精确度 | 较高 | 一般高 |
处理突发流量 | 不支持 | 支持,通过桶最大容量 |
松弛度 | 支持 | 不支持 |
灵活度 | 可以设置是否采用最大松弛度 | 可以设置桶容量三种限流判定方法 wait, allow, reserve |
用哪个好呢?
用哪个好,从来没有标准答案,只有哪种方法最适合哪种场景。
- 请求QPS极高,要求按照时间平均
由于令牌桶的时间计算方式,是根据(时间差*QPS)获取桶的数目,这种float64+带乘法的计算过程,由于是float64形式的计算,高QPS的情况下(例如1000000),会存在一定的时间误差。因此这种情况更推荐采用Uber的包,采用int64的计算方式,且没有乘法操作,相比会有更高一些的精度。
(当然,通过本人实测,如果QPS1000以内,我感觉这两种方法没有本质的差别) - 需要具备一定量的突发流量处理能力
由于漏桶算法和令牌桶算法本质的区别,令牌桶算法能够很好地处理突发流量,因此这种场景推荐使用time/rate,可以通过设置比较大的brust(桶最大值)来保证处理突发流量的能力。 - 严格的限制请求的固定间隔
都可以,其中Uber方法时间把控更为严格,可以通过Uber方法中ratelimit.WithoutSlack参数,取消松弛量的影响。
limiter := ratelimit.New(100, ratelimit.WithoutSlack)
⭐️其他限流
分布式限流,例如Go Redis分布式限流库 https://github.com/go-redis/redis_rate
也有很多限流的包,还没来得及仔细研究,后面会做一个更加详细的对比