免费发帖网站大全,网站建设公司简介范文,找做模型方案去哪个网站,网站热点关键词文章目录 本系列漏桶限流算法uber的漏桶算法使用mutex版本数据结构获取令牌松弛量 atomic版本数据结构获取令牌测试漏桶的松弛量 总结 本系列
开源限流组件分析#xff08;一#xff09;#xff1a;juju/ratelimit开源限流组件分析#xff08;二#xff09;#xff1a;u… 文章目录 本系列漏桶限流算法uber的漏桶算法使用mutex版本数据结构获取令牌松弛量 atomic版本数据结构获取令牌测试漏桶的松弛量 总结 本系列
开源限流组件分析一juju/ratelimit开源限流组件分析二uber-go/ratelimit本文开源限流组件分析三golang-time/rate
漏桶限流算法
漏桶限流算法思路很简单水数据或者请求先进入到漏桶里漏桶以一定的速度出水当水流入速度过大会直接溢出漏桶算法能强行限制请求的处理速度 相比于令牌桶中只要桶内还有剩余令牌调用方就可以马上消费令牌的策略。漏桶相对来说更加严格调用方只能按照预定的间隔顺序消费令牌
例如假设漏桶出水的间隔是10ms上次在0ms时刻出水那么下次是10ms时刻再下次是20ms时刻
uber的漏桶算法
uber对标准的漏桶算法做了一些优化加了一些松弛量这么做了后就能应对突发流量达到和令牌桶一样的效果
关于什么是松弛量在介绍获取令牌流程时再详细分析 阅读的源码https://github.com/uber-go/ratelimit版本v0.3.1 使用
下面展示一个没有使用松弛量的漏桶使用示例
func Example_default() {// 参数100代表每秒放行100个请求即每10ms放行一个请求rl : ratelimit.New(100)prev : time.Now()for i : 0; i 10; i {// Take获取令牌返回获取令牌成功的时间now : rl.Take()if i 0 {// 打印每次放行间隔fmt.Println(i, now.Sub(prev))}prev now}
}打印结果
1 10ms
2 10ms
3 10ms
4 10ms
5 10ms
6 10ms
7 10ms
8 10ms
9 10ms可以看出每次放行间隔10ms符合预期
uber的漏桶提供了mutex版本和atomic版本两种实现主要区别在于怎么控制并发以及怎么记录松弛量 mutex版本
数据结构
type mutexLimiter struct {sync.Mutex// 上次请求放行的时刻last time.Time// 桶中积累的松弛量sleepFor time.Duration// 每个请求之间的间隔perRequest time.Duration// 最大松弛量maxSlack time.Durationclock Clock
}初始化桶
func newMutexBased(rate int, opts ...Option) *mutexLimiter {config : buildConfig(opts)perRequest : config.per / time.Duration(rate)l : mutexLimiter{perRequest: perRequest,// config.slack默认值10 例如当perRequest10ms时// maxSlack就是 -10 * 100ms -1smaxSlack: -1 * time.Duration(config.slack) * perRequest,clock: config.clock,}return l
}func buildConfig(opts []Option) config {c : config{clock: clock.New(),// 最大松弛量默认是perRequest的10倍slack: 10,per: time.Second,}for _, opt : range opts {opt.apply(c)}return c
}主要设置两个变量 每次放行时间间隔perRequest根据参数rate计算rate表示每单位时间per能放行多少请求 例如当per1srate100时每次放行时间间隔perRequest10ms 最大松弛量maxSlackconfig.slack默认值10 例如当perRequest10ms时maxSlack就是 -10 * 100ms -1s 获取令牌
要实现标准的漏桶算法其实比较简单
记录上次放行时间last当本次请求到来时, 如果此刻的时间与last相比并没有达到 perRequest 规定的间隔大小 sleep 一段时间即可 对应代码为删除了松弛相关代码
func (t *mutexLimiter) Take() time.Time {t.Lock()defer t.Unlock()now : t.clock.Now()// 如果第一次请求直接放行if t.last.IsZero() {t.last nowreturn t.last}// 当前时间距离上次放行时间的间隔sleepFor : t.perRequest - now.Sub(t.last)// 如果比规定的间隔小需要sleepif sleepFor 0 {t.clock.Sleep(t.sleepFor)t.last now.Add(t.sleepFor)t.sleepFor 0return t.last
}松弛量
如果不引入松弛量按照标准漏桶算法的流程
假设现在有三个请求req1、req2 和 req3获取令牌间隔perRequest规定为 10ms req1 先到来 当 req1 完成之后 20ms req2 才到来此时距离上次放行有20ms大于10ms可以对req2 放行 当 req2 放行后5ms req3 到来此时距离上次放行才5ms不足 规定的间隔10ms因此还需要等待 5ms 才能继续放行req3 这种策略有什么问题无法应对任何突发流量没有弹性定死了相邻两次请求放行的间隔必须大于等于规定的值
于是引入了松弛量的概念当较长时间超过perRequest没有请求到来时会在桶中积累一些松弛量这样接下来的请求可以先消耗松弛量而不会被漏洞的获取令牌间隔限制。直到把松弛量耗尽为止
当然松弛量也不是无限积累这样当很长时间没有请求后就跟没有限流没区别了。因此桶中规定了最多积累多少松弛量maxSlack
加上松弛量后代获取令牌代码如下 计算perRequest - now.Sub(t.last) 如果大于0说明当前时间距离上次放行时间不足规定间隔需要等待或者需要消耗松弛量如果小于0说明当前时间距离上次放行时间超过了规定间隔那么当前请求一定可以放行并且可以在桶中积累一些松弛量给下次请求使用将结果累加到t.sleepFor 假设累加的松弛值超过了maxSlack修正为maxSlack默认为10倍的规定间隔 如果t.sleepFor 0说明本次请求应该等待的时间在使用完桶中的松弛量后还不够需要sleep这段时间结束后将sleepFor置为0 否则t.sleepFor 0说明桶中的存储松弛余量可以满足本次请求本次请求放行
func (t *mutexLimiter) Take() time.Time {t.Lock()defer t.Unlock()now : t.clock.Now()// 如果第一次请求直接放行if t.last.IsZero() {t.last nowreturn t.last}// 假设perRequest10msnow.Sub(t.last)5ms那么需要睡5ms// 假设perRequest10msnow.Sub(t.last)15ms说明距离上传请求的时间超过了漏桶的限流间隔10ms那么sleepFor变为-5t.sleepFor t.perRequest - now.Sub(t.last)// 默认最多松弛 10倍的perRequest// 如果累加的松弛值超过了maxSlack修正为maxSlackif t.sleepFor t.maxSlack {t.sleepFor t.maxSlack}// 需要sleepif t.sleepFor 0 {t.clock.Sleep(t.sleepFor)t.last now.Add(t.sleepFor)t.sleepFor 0// 如果sleepFor 0说明还有存储的松弛余量可以放行} else {t.last now}return t.last
}atomic版本
数据结构
type atomicInt64Limiter struct {// 上次在哪个时刻发放许可state int64// 每次请求间隔perRequest time.Duration// 最大松弛量maxSlack time.Durationclock Clock
}func newAtomicInt64Based(rate int, opts ...Option) *atomicInt64Limiter {config : buildConfig(opts)perRequest : config.per / time.Duration(rate)l : atomicInt64Limiter{perRequest: perRequest,// 最大松弛量10倍的perRequestmaxSlack: time.Duration(config.slack) * perRequest,clock: config.clock,}atomic.StoreInt64(l.state, 0)return l
}获取令牌
atomic版本的漏桶不是用一个变量存储桶中的松弛量而是记录了上次在哪个时刻发放许可timeOfNextPermissionIssue那松弛量用什么表示呢当前时间now减去上次发放许可时间timeOfNextPermissionIssue就代表松弛量
怎么理解这个值代表应该在什么时刻发放令牌
如果该值 比now大说明需要sleep一段时间等待时间流逝本次才能获取到令牌如果该值 比now小说明应该在更早的时间就可以获取令牌本次请求不需要等待
每次请求时会在timeOfNextPermissionIssue的基础上加上perRequest并更新timeOfNextPermissionIssue表示本次发放许可的时间比上次前进了perRequest时间
如果较长时间没有请求那天然就积累了一些松弛量因为上次发放许可时间就会比较小即便加上本次的消耗perRequest也没有now大就可以直接放行 如果上次发放许可时间加上固定间隔perRequest小于now说明本次可以放行并更新上次发放许可时间 perRequest 当然timeOfNextPermissionIssue不能比now小太多否则限流就没意义了。因此规定该值最多比now小10倍的perRequest表示松弛量最多为perRequest的10倍 否则不能放行需要等待时间流逝到timeOfNextPermissionIssue perRequest为止
func (t *atomicInt64Limiter) Take() time.Time {var (newTimeOfNextPermissionIssue int64now int64)for {now t.clock.Now().UnixNano()// 上次在哪个时刻发放许可timeOfNextPermissionIssue : atomic.LoadInt64(t.state)switch {// 第一次请求或者不是第一次请求且没有松弛量且now 比 上次发放许可时间 多了超过perRequest case timeOfNextPermissionIssue 0 || (t.maxSlack 0 now-timeOfNextPermissionIssue int64(t.perRequest)):// 本次放行本次在now时刻发放许可newTimeOfNextPermissionIssue now// 可以有松弛量且 当前时间 - 上次发放许可时间 11倍perRequestcase t.maxSlack 0 now-timeOfNextPermissionIssue int64(t.maxSlack)int64(t.perRequest):// 本次在 now - 最大松弛量 时刻发放许可newTimeOfNextPermissionIssue now - int64(t.maxSlack)/**1.没有松弛量且当前时间距离 上次发放许可时间 超过了 不到perRequest本次应该在上次发放许可时间 perRequest 再发放2.有松弛量且 当前时间 - 上次发放许可时间 11倍perRequest那本次应该在上次发放许可时间 perRequest 再发放*/default:// 更新上次发放许可时间 perRequestnewTimeOfNextPermissionIssue timeOfNextPermissionIssue int64(t.perRequest)}if atomic.CompareAndSwapInt64(t.state, timeOfNextPermissionIssue, newTimeOfNextPermissionIssue) {break}}sleepDuration : time.Duration(newTimeOfNextPermissionIssue - now)// 没有余量了需要sleepif sleepDuration 0 {t.clock.Sleep(sleepDuration)// 返回放行时的时间return time.Unix(0, newTimeOfNextPermissionIssue)}return time.Unix(0, now)
}测试漏桶的松弛量
下面对漏桶的松弛量进行测试如果先积累一些松弛量就会起到因对突发流量的效果例如让时间先走45ms
func Example_default() {// 每10ms放行一个请求, 最大松弛量100msrl : ratelimit.New(100)// 在0ms发放许可rl.Take()// 此时时间来到45mstime.Sleep(time.Millisecond * 45)prev : time.Now()for i : 0; i 10; i {now : rl.Take()if i 0 {fmt.Println(i, now.Sub(prev))}prev now}
}打印结果
1 1µs
2 103µs
3 4µs
4 4.79ms
5 10ms
6 10ms
7 10ms
8 10ms
9 10ms可以看出前4次都没等直接获取到令牌第4次等了大约5ms之后每次放行间隔都是10ms
执行到for循环时上次发放许可时间10ms当前时刻45ms
拆解for循环中的每次take中的漏桶状态
i0, 下次应该在10ms发放许可当前时刻45ms放行i1, 下次应该在20ms发放许可当前时刻45ms放行i2, 下次应该在30ms发放许可当前时刻45ms放行i3, 下次应该在40ms发放许可当前时刻45ms放行i4, 下次应该在50ms发放许可当前时刻45ms更新timeOfNextPermissionIssue50需要sleep 5msi5, 下次应该在60ms发放许可当前50ms需要sleep10ms更新timeOfNextPermissionIssue60
如下图所示 总结
加上松弛量后这个漏桶和标准的令牌桶区别就不大了
都能应对一定的突发流量当桶中没有松弛量没有令牌时都按照固定时间间隔放行请求
这个库有个问题是对桶中等待中的请求数没有限制这样当并发量非常大时会导致请求一直在桶中堆积
因此工程中要使用的话最好对源码进行改造当等待中的请求数超过阈值就快速返回失败
