Go 基于令牌桶的限流器实现

简介

如果一般流量过大，下游系统反应不过来，这个时候就需要限流了，其实和上地铁是一样的，就是减慢上游访问下游的速度。
限制访问服务的频次或者频率，防止服务过载，被刷爆等。

golang 官方扩展包 time(golang.org/x/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*31), 2)
 //time.sleep(time.second)
 for i := 0; i < 10; i++ {
  var ok bool
  if limiter.allow() {
   ok = true
  }
  time.sleep(time.millisecond * 20)
  fmt.println(ok, limiter.burst())
 }
}

执行结果：
=== run   testlimter
true 2
true 2
true 2
false 2
true 2
true 2
false 2
true 2
true 2
false 2
--- pass: testlimter (0.21s)

通过执行结果可以看到， 令牌桶开始是2个满的，由于令牌的间隔比请求的间隔多了11ms（31-20）, 所以每两个请求会失败一次。

具体实现原理

先看下限流器的创建方法： newlimiter

func newlimiter(r limit, b int) *limiter {
 return &limiter{
  limit: r,
  burst: b,
 }
}

查看限流器数据结构 limiter

// the methods allown, reserven, and waitn consume n tokens.
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
}

• burst 表示了桶的大小
• limit 表示放入桶的频率
• tokens 表示剩余令牌个数
• last 最近取走 token 的时间
• lastevent 最近限流事件的时间

当令牌桶发放后，会保留在 reservation 对象中， 定义如下， reservation 对象，描述了一个达到 timetoact 时间后，可以获取到的令牌的数量 tokens 数。

type reservation struct {
  ok        bool  // 是否满足条件分配到了tokens
  lim       *limiter // 发送令牌的限流器
  tokens    int   // tokens 的数量
  timetoact time.time  //  满足令牌发放的时间
  limit limit  // 令牌发放速度
}

限流器如何限流

官方提供的限流器有阻塞等待， 也有直接判断方式的， 还有提供维护预留式等。
如何实现限流的代码，在 reserven 中。

使用时，每次都调用了 allow() 方法

// allow is shorthand for allown(time.now(), 1).
func (lim *limiter) allow() bool {
 return lim.allown(time.now(), 1)
}

// allown reports whether n events may happen at time now.
// use this method if you intend to drop / skip events that exceed the rate limit.
// otherwise use reserve or wait.
func (lim *limiter) allown(now time.time, n int) bool {
 return lim.reserven(now, n, 0).ok
}

继续查看 reservern  算法

方法说明：

• 三个参数： now, n, maxfuturereserve
• 在 now 时间需要拿到 n 个令牌，最多等待的时间为 maxfuturereserve
• 结果将返回一个预留令牌的对象 reservation

// maxfuturereserve specifies the maximum reservation wait duration allowed.
// reserven returns reservation, not *reservation, to avoid allocation in allown and waitn.
func (lim *limiter) reserven(now time.time, n int, maxfuturereserve time.duration) reservation {
 lim.mu.lock()
 // 首先判断是否放入频次是否为无穷大，如果为无穷大，说明暂时不限流
 if lim.limit == inf {
  lim.mu.unlock()
  return reservation{
   ok:        true,
   lim:       lim,
   tokens:    n,
   timetoact: now,
  }
 }
   // 拿到截止 now 时间时，可以获取的令牌 tokens 数量，上一次拿走令牌的时间是last
 now, last, tokens := lim.advance(now)

 // calculate the remaining number of tokens resulting from the request.
 //  更新 tokens数量，把需要拿走的去掉
 tokens -= float64(n)

 // calculate the wait duration
 // 如果 tokens 数量为负数，说明需要等待，计算等待时间 waitduration
 var waitduration time.duration
 if tokens < 0 {
  waitduration = lim.limit.durationfromtokens(-tokens)
 }

 // decide result
 // 计算是否满足分配要求
 // 1. 需要分配的大小不超过桶容量
 // 2. 等待时间不超过设定的等待时长
 ok := n <= lim.burst && waitduration <= maxfuturereserve

 // prepare reservation
 // 最后构造一个 resvervation 对象
 r := reservation{
  ok:    ok,
  lim:   lim,
  limit: lim.limit,
 }
 if ok {
  r.tokens = n
  r.timetoact = now.add(waitduration)
 }

 // update state
   // 需要更新当前 limit 的值 
 if ok {
  lim.last = now
  lim.tokens = tokens
  lim.lastevent = r.timetoact
 } else {
  lim.last = last
 }

 lim.mu.unlock()
 return r
}

从实现上看， limiter 并不是每隔一段时间更新当前桶的数量，而是记录了上次访问时和当前桶中令牌的数量，当再次访问时，通过上次访问时间计算出当前令牌的数量，决定是否可以发放令牌。

到此这篇关于go 基于令牌桶的限流器实现的文章就介绍到这了,更多相关go 令牌桶限流器内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！