Go 限流器-漏桶 VS 令牌桶常用包原理解析

🕗 发布于 2024-03-24 23:45 golang oracle 算法 数据结构

本文主要介绍两个包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 second

    prev := 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 = now
       case 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 main

import (
    "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 updated
    last 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.burst
    limit := lim.limit
    lim.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 := InfDuration
    if 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 testing
    select {
    case <-ch:
       return nil
    case <-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 bool
       if lim.burst >= n {
          ok = true
          lim.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.Duration
    if tokens < 0 {
       waitDuration = lim.limit.durationFromTokens(-tokens)
    }

    // 边界校验，是否预占成功
    ok := n <= lim.burst && waitDuration <= maxFutureReserve

    r := Reservation{
       ok:    ok,
       lim:   lim,
       limit: lim.limit,
    }
    if ok {
       r.tokens = n
       r.timeToAct = t.Add(waitDuration)  // 需要等待的时间

       // Update state
       lim.last = t
       lim.tokens = tokens
       lim.lastEvent = r.timeToAct
    }

    return r
}

advance计算得到此刻桶内tokens数

func (lim *Limiter) advance(t time.Time) (newT time.Time, newTokens float64) {
    last := lim.last
    if t.Before(last) {
       last = t
    }

    // （（当前时间 - 上次请求时间） * 每秒产生的tokens数量）+ 桶内原有tokens数 = 当前桶内tokens数
    elapsed := t.Sub(last)
    delta := lim.limit.tokensFromDuration(elapsed)
    tokens := lim.tokens + delta
    if 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

也有很多限流的包，还没来得及仔细研究，后面会做一个更加详细的对比

原文地址：https://blog.csdn.net/t949500898/article/details/136910177

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：apifox创建接口含中文字符报错的两种解决方案
下一篇：UE5制作推箱子动作时获取物体与角色朝向的角度及跨蓝图修改变量

学生公寓管理系统
由于某些特殊原因，需要对学生当前入住的床位进行修改时，使用此功能。新生开学后，宿管管理可以先根据学生所在系查询该专业预定的宿舍和具体的床位入住状态信息，为该学生分配一个空闲状态的床位，并修改当前床位状
阅读更多2024-11-15
Windows上安装RabbitMQ 4.0.3教程
通过上述步骤，你应该已经成功在Windows上安装了RabbitMQ 4.0.3，并启用了管理插件。现在，你可以开始使用RabbitMQ来构建你的分布式系统了。希望这篇教程对你有所帮助，如果有任何问题
阅读更多2024-11-15
【蓝桥等考C++真题】蓝桥杯等级考试C++组第13级L13真题原题(含答案)-成绩排序
C++L13 成绩排序(50 分)
阅读更多2024-11-15
Web入门
HTTP（HyperText Transfer Protocol，超文本传输协议）是一种应用层协议。主要用于在Web浏览器和服务器之间传输数据。基于TCP协议：面向连接，确保了数据传输的可靠性和顺序。
阅读更多2024-11-15
Android 最新的AndroidStudio引入依赖失败如何解决？如:Failed to resolve:xxxx
在引入依赖时报错：Failed to resolve: xxx.xxxx:1.1.0。
阅读更多2024-11-15
力扣113：路径总和II
力扣113：路径总和II。C语言
阅读更多2024-11-15
【springboot使用sqlite数据库】Java后台同时使用mysql、sqlite
根据业务的需要，老版程序使用的数据库是sqlite，版本升级成前后台分离模式，因此需要兼容mysql与sqlite数据库同时使用。
阅读更多2024-11-15
力扣.15 三数之和 three-sum
这道题作为 leetcode 的第 15 道题，看起来似曾相识。大概思路可以有下面几种：暴力解法数组排序+二分Hash 优化双指针。
阅读更多2024-11-15
力扣第 53 题：最大子数组和
题目给定一个整数数组nums，目标是找到和最大的连续子数组（最少包含一个元素），并返回其和。例如，在数组中，和最大的子数组为，其和为6。
阅读更多2024-11-15
linux-缓冲区
调用系统调用接口是需要时间成本的，如果使用缓冲区，存储的字符串到一定长度再调用系统调用，能够有效减少调用系统调用的次数。这个缓冲区是语言设置的，缓冲区满足条件就用系统调用接口write将缓冲区的内容刷
阅读更多2024-11-15

Go 限流器-漏桶 VS 令牌桶 常用包原理解析

背景

使用场景：

限流器通用实现方案

Uber “改良版”的漏桶限流

1. 使用方法

1.1 初始化

1.2 限流

1.3 举个🌰

2. 原理分析

2.1 初始化New方法

2.2 atomicInt64Limiter 对象

2.3 限流Take 方法

3. Case分析

time/rate令牌桶Plus

1. 使用方法

1.1 初始化

1.2 限流

1.3 举个🌰

2. 原理分析

2.1 初始化NewLimiter方法

2.2 限流方法

3. Case分析

对比一下

共同点

不同点

用哪个好呢？

⭐️其他限流

相关文章

Go 限流器-漏桶 VS 令牌桶常用包原理解析