漏桶、令牌桶限流的Go语言实现

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文首发于我的个人博客：liwenzhou.com，更多更详细的Go语言项目实战内容就在liwenzhou.com。

限流

限流又称为流量控制（流控），通常是指限制到达系统的并发请求数。

我们生活中也会经常遇到限流的场景，比如：某景区限制每日进入景区的游客数量为8万人；沙河地铁站早高峰通过站外排队逐一放行的方式限制同一时间进入车站的旅客数量等。

限流虽然会影响部分用户的使用体验，但是却能在一定程度上报障系统的稳定性，不至于崩溃（大家都没了用户体验）。

而互联网上类似需要限流的业务场景也有很多，比如电商系统的秒杀、微博上突发热点新闻、双十一购物节、12306抢票等等。这些场景下的用户请求量通常会激增，远远超过平时正常的请求量，此时如果不加任何限制很容易就会将后端服务打垮，影响服务的稳定性。

此外，一些厂商公开的API服务通常也会限制用户的请求次数，比如百度地图开放平台等会根据用户的付费情况来限制用户的请求数等。

常用的限流策略

漏桶

漏桶法限流很好理解，假设我们有一个水桶按固定的速率向下方滴落一滴水，无论有多少请求，请求的速率有多大，都按照固定的速率流出，对应到系统中就是按照固定的速率处理请求。

漏桶算法原理

漏桶法的关键点在于漏桶始终按照固定的速率运行，但是它并不能很好的处理有大量突发请求的场景，毕竟在某些场景下我们可能需要提高系统的处理效率，而不是一味的按照固定速率处理请求。

关于漏桶的实现，uber团队有一个开源的github.com/uber-go/ratelimit库。这个库的使用方法比较简单，Take() 方法会返回漏桶下一次滴水的时间。

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 0
    // 1 10ms
    // 2 10ms
    // 3 10ms
    // 4 10ms
    // 5 10ms
    // 6 10ms
    // 7 10ms
    // 8 10ms
    // 9 10ms
}

它的源码实现也比较简单，这里大致说一下关键的地方，有兴趣的同学可以自己去看一下完整的源码。

限制器是一个接口类型，其要求实现一个Take()方法：

type Limiter interface {
 // Take方法应该阻塞已确保满足 RPS
 Take() time.Time
}

实现限制器接口的结构体定义如下，这里可以重点留意下maxSlack字段，它在后面的Take()方法中的处理。

type limiter struct {
 sync.Mutex                // 锁
 last       time.Time      // 上一次的时刻
 sleepFor   time.Duration  // 需要等待的时间
 perRequest time.Duration  // 每次的时间间隔
 maxSlack   time.Duration  // 最大的富余量
 clock      Clock          // 时钟
}

limiter结构体实现Limiter接口的Take()方法内容如下：

// Take 会阻塞确保两次请求之间的时间走完
// Take 调用平均数为 time.Second/rate.
func (t *limiter) Take() time.Time {
 t.Lock()
 defer t.Unlock()

 now := t.clock.Now()

 // 如果是第一次请求就直接放行
 if t.last.IsZero() {
  t.last = now
  return t.last
 }

 // sleepFor 根据 perRequest 和上一次请求的时刻计算应该sleep的时间
 // 由于每次请求间隔的时间可能会超过perRequest, 所以这个数字可能为负数，并在多个请求之间累加
 t.sleepFor += t.perRequest - now.Sub(t.last)

 // 我们不应该让sleepFor负的太多，因为这意味着一个服务在短时间内慢了很多随后会得到更高的RPS。
 if t.sleepFor < t.maxSlack {
  t.sleepFor = t.maxSlack
 }

 // 如果 sleepFor 是正值那么就 sleep
 if t.sleepFor > 0 {
  t.clock.Sleep(t.sleepFor)
  t.last = now.Add(t.sleepFor)
  t.sleepFor = 0
 } else {
  t.last = now
 }

 return t.last
}

上面的代码根据记录每次请求的间隔时间和上一次请求的时刻来计算当次请求需要阻塞的时间——sleepFor，这里需要留意的是sleepFor的值可能为负，在经过间隔时间长的两次访问之后会导致随后大量的请求被放行，所以代码中针对这个场景有专门的优化处理。创建限制器的New()函数中会为maxSlack设置初始值，也可以通过WithoutSlack这个Option取消这个默认值。

func New(rate int, opts ...Option) Limiter {
 l := &limiter{
  perRequest: time.Second / time.Duration(rate),
  maxSlack:   -10 * time.Second / time.Duration(rate),
 }
 for _, opt := range opts {
  opt(l)
 }
 if l.clock == nil {
  l.clock = clock.New()
 }
 return l
}

令牌桶

令牌桶其实和漏桶的原理类似，令牌桶按固定的速率往桶里放入令牌，并且只要能从桶里取出令牌就能通过，令牌桶支持突发流量的快速处理。

令牌桶原理

对于从桶里取不到令牌的场景，我们可以选择等待也可以直接拒绝并返回。

对于令牌桶的Go语言实现，大家可以参照github.com/juju/ratelimit库。这个库支持多种令牌桶模式，并且使用起来也比较简单。

创建令牌桶的方法：

// 创建指定填充速率和容量大小的令牌桶
func NewBucket(fillInterval time.Duration, capacity int64) *Bucket
// 创建指定填充速率、容量大小和每次填充的令牌数的令牌桶
func NewBucketWithQuantum(fillInterval time.Duration, capacity, quantum int64) *Bucket
// 创建填充速度为指定速率和容量大小的令牌桶
// NewBucketWithRate(0.1, 200) 表示每秒填充20个令牌
func NewBucketWithRate(rate float64, capacity int64) *Bucket

取出令牌的方法如下：

// 取token（非阻塞）
func (tb *Bucket) Take(count int64) time.Duration
func (tb *Bucket) TakeAvailable(count int64) int64

// 最多等maxWait时间取token
func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool)

// 取token（阻塞）
func (tb *Bucket) Wait(count int64)
func (tb *Bucket) WaitMaxDuration(count int64, maxWait time.Duration) bool

虽说是令牌桶，但是我们没有必要真的去生成令牌放到桶里，我们只需要每次来取令牌的时候计算一下，当前是否有足够的令牌就可以了，具体的计算方式可以总结为下面的公式：

当前令牌数 = 上一次剩余的令牌数 + (本次取令牌的时刻-上一次取令牌的时刻)/放置令牌的时间间隔 * 每次放置的令牌数

github.com/juju/ratelimit这个库中关于令牌数计算的源代码如下：

func (tb *Bucket) currentTick(now time.Time) int64 {
 return int64(now.Sub(tb.startTime) / tb.fillInterval)
}

func (tb *Bucket) adjustavailableTokens(tick int64) {
 if tb.availableTokens >= tb.capacity {
  return
 }
 tb.availableTokens += (tick - tb.latestTick) * tb.quantum
 if tb.availableTokens > tb.capacity {
  tb.availableTokens = tb.capacity
 }
 tb.latestTick = tick
 return
}

获取令牌的TakeAvailable()函数关键部分的源代码如下：

func (tb *Bucket) takeAvailable(now time.Time, count int64) int64 {
 if count <= 0 {
  return 0
 }
 tb.adjustavailableTokens(tb.currentTick(now))
 if tb.availableTokens <= 0 {
  return 0
 }
 if count > tb.availableTokens {
  count = tb.availableTokens
 }
 tb.availableTokens -= count
 return count
}

大家从代码中也可以看到其实令牌桶的实现并没有很复杂。

gin框架中使用限流中间件

在gin框架构建的项目中，我们可以将限流组件定义成中间件。

这里使用令牌桶作为限流策略，编写一个限流中间件如下：

func RateLimitMiddleware(fillInterval time.Duration, cap int64) func(c *gin.Context) {
 bucket := ratelimit.NewBucket(fillInterval, cap)
 return func(c *gin.Context) {
  // 如果取不到令牌就中断本次请求返回 rate limit...
  if bucket.TakeAvailable(1) < 1 {
   c.String(http.StatusOK, "rate limit...")
   c.Abort()
   return
  }
  c.Next()
 }
}