Golang

关注公众号 jb51net

关闭
首页 > 脚本专栏 > Golang > golang rate令牌桶源码分析实现

golang rate令牌桶源码分析实现方式

作者:dandyhuang

这篇文章主要介绍了golang rate令牌桶源码分析实现方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教

golang rate令牌桶实现

高并发三板斧:

限流、缓存、降级。

限流其实就是:

当高并发或者瞬时高并发时,为了保证系统的稳定性、可用性,系统以牺牲部分请求为代价或者延迟处理请求为代价,保证系统整体服务可用。

今天就给大家介绍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参数中

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亲缘性问题就越明显。

个人觉得虽然加锁的实现,对性能有一部分影响,但是令牌桶都是在计算,所以性能不会有很大的问题,加锁时间不长。

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文