上一篇說到 Leaky Bucket 能限制客戶端的訪問速率, 但是無法應(yīng)對(duì)突發(fā)流量, 本質(zhì)原因就是漏斗桶只是為了保證固定時(shí)間內(nèi)通過的流量是一樣的. 面對(duì)這種情況, 本篇文章繼續(xù)介紹另外一種限流器: Token Bucket -- 令牌桶

什么是 Token Bucket

漏斗桶的桶空間就那么大, 其只能保證桶里的請(qǐng)求是勻速流出, 并不關(guān)心流入的速度, 只要桶溢出了就服務(wù)拒絕, 這可能并不符合互聯(lián)網(wǎng)行業(yè)的使用場(chǎng)景.

試想這樣的場(chǎng)景, 盡管服務(wù)器的 QPS 已經(jīng)達(dá)到限速閾值了, 但是并不想將所有的流量都拒之門外, 仍然讓部分流量能夠正常通過限流器. 這樣我們的服務(wù)器在面對(duì)突發(fā)流量時(shí)能夠有一定的伸縮空間, 而不是一直處于不可用狀態(tài).

基于上面的場(chǎng)景需求, 令牌桶采用跟漏斗桶截然不同的做法.

令牌桶

令牌桶也有自己的固定大小, 我們?cè)O(shè)置 QPS 為 100, 在初始化令牌桶的時(shí)候, 就會(huì)立即生成 100 個(gè)令牌放到桶里面, 同時(shí)還按照一定的速率, 每隔一定的時(shí)間產(chǎn)生固定數(shù)量的令牌放入到桶中. 如果桶溢出了, 則舍棄生成的令牌.

只要有請(qǐng)求能夠拿到令牌, 就能保證其通過限流器. 當(dāng)然拿不到令牌的請(qǐng)求只能被無情拒絕了(或者等待令牌產(chǎn)生), 這個(gè)請(qǐng)求的命不好~

面對(duì)突然爆發(fā)的流量, 可能大部分流量都被限流器給擋住了, 但是也有部分流量剛好拿到了剛剛生成的 Token, 就這樣在千軍萬馬中通過了限流器. 相對(duì)于漏斗桶來說, 令牌桶更適合現(xiàn)在互聯(lián)網(wǎng)行業(yè)的需要, 是被廣泛使用的限流算法

如何設(shè)置令牌桶的大小和產(chǎn)生令牌的速率?

答: 多進(jìn)行生產(chǎn)環(huán)境的壓測(cè), 根據(jù)集群的實(shí)際承受能力設(shè)置相應(yīng)桶的大小和產(chǎn)生令牌的速率. 血的經(jīng)驗(yàn)告訴我, 周期性的線上壓測(cè)是一件很重要的事情(使用local cache的程序, 壓測(cè)的時(shí)候一定要記得先臨時(shí)關(guān)閉它)

juju/ratelimit

juju/ratelimit 是大部分項(xiàng)目都在使用的 golang 令牌桶的實(shí)現(xiàn)方案. 下面會(huì)結(jié)合其用法, 源碼剖析令牌桶的實(shí)現(xiàn)的方案.

gin 中間件

package?main

import?(
?"fmt"
?"log"
?"net/http"
?"time"

?"github.com/gin-gonic/gin"
?"github.com/juju/ratelimit"
)

var?limiter?=?ratelimit.NewBucketWithQuantum(time.Second,?10,?10)

func?tokenRateLimiter()?gin.HandlerFunc?{
?fmt.Println("token?create?rate:",?limiter.Rate())
?fmt.Println("available?token?:",?limiter.Available())
?return?func(context?*gin.Context)?{
??if?limiter.TakeAvailable(1)?==?0?{
???log.Printf("available?token?:%d",?limiter.Available())
???context.AbortWithStatusJSON(http.StatusTooManyRequests,?"Too?Many?Request")
??}?else?{
???context.Writer.Header().Set("X-RateLimit-Remaining",?fmt.Sprintf("%d",?limiter.Available()))
???context.Writer.Header().Set("X-RateLimit-Limit",?fmt.Sprintf("%d",?limiter.Capacity()))
???context.Next()
??}
?}
}

func?main()?{
?e?:=?gin.Default()
?e.GET("/test",?tokenRateLimiter(),?func(context?*gin.Context)?{
??context.JSON(200,?true)
?})
?e.Run(":9091")
}

Output:

token?create?rate:?100
available?token?:?100
[GIN]?2020/07/03?-?17:34:37?|?200?|?????157.505μs?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:34:37?|?200?|?????310.898μs?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:34:37?|?200?|???????61.64μs?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:34:37?|?200?|???????8.677μs?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:34:37?|?200?|???????6.145μs?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:34:37?|?200?|??????23.576μs?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:34:37?|?200?|???????5.617μs?|???????127.0.0.1?|?GET??????/test
.....
[GIN]?2020/07/03?-?17:35:03?|?429?|????6.193792ms?|???????127.0.0.1?|?GET??????/test
[GIN]?2020/07/03?-?17:35:03?|?200?|???????8.509μs?|???????127.0.0.1?|?GET??????/test?[GIN]?2020/07/03?-?17:35:03?|?429?|??????10.324μs?|???????127.0.0.1?|?GET??????/test
....

有下面特點(diǎn):

令牌桶初始化后里面就有 100 個(gè)令牌

每秒鐘會(huì)產(chǎn)生 100 個(gè)令牌, 保證每秒最多有 100 個(gè)請(qǐng)求通過限流器, 也就是說 QPS 的上限是 100

流量過大時(shí)能夠啟動(dòng)限流, 在限流過程中, 仍然能讓部分流量通過

源碼分析

初始化

建議使用初始化函數(shù)有下面三種:

• NewBucket(fillInterval time.Duration, capacity int64): 默認(rèn)的初始化函數(shù), 每一個(gè)周期內(nèi)生成 1 個(gè)令牌, 默認(rèn) quantum = 1
• NewBucketWithQuantum(fillInterval time.Duration, capacity, quantum int64) : 跟 NewBucket 類似, 每個(gè)周期內(nèi)生成 quantum 個(gè)令牌
• NewBucketWithRate(rate float64, capacity int64): 每秒產(chǎn)生 rate 速率的令牌

其實(shí)初始化函數(shù)還有很多種, 但基本上大同小異, 最后都是調(diào)用 NewBucketWithQuantumAndClock.

func?NewBucketWithQuantumAndClock(fillInterval?time.Duration,?capacity,?quantum?int64,?clock?Clock)?*Bucket?{
?//?....
?return?&Bucket{
??clock:???????????clock,
??startTime:???????clock.Now(),
??latestTick:??????0,????????????//?上一次產(chǎn)生token的記錄點(diǎn)
??fillInterval:????fillInterval,?//?產(chǎn)生token的間隔
??capacity:????????capacity,?????//?桶的大小
??quantum:?????????quantum,??????//?每秒產(chǎn)生token的速率
??availableTokens:?capacity,?????//?桶內(nèi)可用的令牌的個(gè)數(shù)
?}
}

Rate

func?(tb?*Bucket)?Rate()?float64?{
?return?1e9?*?float64(tb.quantum)?/?float64(tb.fillInterval)
}

time.Duration 實(shí)際的是以 nanosecond 試試呈現(xiàn)的, 就是 1e9, 1e9 / float64(tb.fillInterval) 的結(jié)果就是 1/tb.fillInterval 秒.

于是令牌桶產(chǎn)生令牌的速率是: 每秒內(nèi)產(chǎn)生 float64(tb.quantum) / float64(tb.fillInterval)

TakeAvailable

func?(tb?*Bucket)?currentTick(now?time.Time)?int64?{
?//?由于?tb.quantum?是每秒產(chǎn)生的token的數(shù)量.?于是計(jì)算從bucket初始化的startTime到現(xiàn)在now的時(shí)間間隔?t,
?//?t/tb.fillInterval?*?tb.quantum?計(jì)算的是從開始到現(xiàn)在應(yīng)該產(chǎn)生的?token?數(shù)量
?return?int64(now.Sub(tb.startTime)?/?tb.fillInterval)
}

func?(tb?*Bucket)?adjustavailableTokens(tick?int64)?{
?if?tb.availableTokens?>=?tb.capacity?{?//?如果令牌的可用數(shù)量已經(jīng)達(dá)到桶的容量,?直接返回
??return
?}
?
?//?tick?*?tb.quantum?是從bucket初始化到本次請(qǐng)求應(yīng)該產(chǎn)生的?token的數(shù)量
?//?tb.latestTick?是從bucket初始化到上次請(qǐng)求應(yīng)該產(chǎn)生的?token的數(shù)量
?//?tick?*?tb.quantum?-?tb.latestTick?計(jì)算出兩次請(qǐng)求間應(yīng)該產(chǎn)生的token數(shù)量
?//?tb.availableTokens?+=?(tick?-?tb.latestTick)?*?tb.quantum:?桶內(nèi)剩余的token數(shù)量?+?新產(chǎn)生的token數(shù)量
?tb.availableTokens?+=?(tick?-?tb.latestTick)?*?tb.quantum
?if?tb.availableTokens?>?tb.capacity?{
??tb.availableTokens?=?tb.capacity?//?如果產(chǎn)生的令牌數(shù)量超過了桶的容量,?則桶內(nèi)剩余的令牌數(shù)量等于桶的size
?}
?tb.latestTick?=?tick
?return
}

func?(tb?*Bucket)?takeAvailable(now?time.Time,?count?int64)?int64?{
?if?count?<=?0?{
??return?0
?}
?tb.adjustavailableTokens(tb.currentTick(now))
?if?tb.availableTokens?<=?0?{?//?如果桶內(nèi)剩余token數(shù)量小于等于0,?直接返回0
??return?0
?}
?if?count?>?tb.availableTokens?{
??count?=?tb.availableTokens
?}
?tb.availableTokens?-=?count
?return?count
}

//?如果返回值是0,?代表桶內(nèi)已經(jīng)沒有令牌了
func?(tb?*Bucket)?TakeAvailable(count?int64)?int64?{
?tb.mu.Lock()
?defer?tb.mu.Unlock()
?return?tb.takeAvailable(tb.clock.Now(),?count)
}

TakeAvailable 是 Token Bucket的核心函數(shù). 從這個(gè)實(shí)現(xiàn)我們能看到

jujue/ratelimit 計(jì)算出請(qǐng)求間隔中應(yīng)該產(chǎn)生的token的數(shù)量, 并不是另外啟動(dòng)一個(gè) goroutine 專門定時(shí)產(chǎn)生固定數(shù)量的token

桶內(nèi)令牌在產(chǎn)生過程中是累加的, 同時(shí)減去每次調(diào)用消耗的數(shù)量

初始化后桶內(nèi)的令牌數(shù)量就是桶的大小

示意圖

Token Bucket的缺陷

令牌桶算法能滿足絕大部分服務(wù)器限流的需要, 是被廣泛使用的限流算法, 不過其也有一些缺點(diǎn):

令牌桶是沒有優(yōu)先級(jí)的，無法讓重要的請(qǐng)求先通過

OP可能因?yàn)橛布收先フ{(diào)整資源, 系統(tǒng)負(fù)載也會(huì)隨著在變化, 如果對(duì)服務(wù)限流進(jìn)行縮容和擴(kuò)容，需要人為手動(dòng)去修改，運(yùn)維成本比較大

令牌桶只能對(duì)局部服務(wù)端的限流, 無法掌控全局資源

下一篇我們看alibaba/Sentinel, kratos 的 bbr 算法是如何做到系統(tǒng)自適應(yīng)限流

參考

[1] 分布式服務(wù)限流實(shí)戰(zhàn)，已經(jīng)為你排好坑了 https://www.infoq.cn/article/Qg2tX8fyw5Vt-f3HH673)

[2] 維基百科--Token_bucket https://en.wikipedia.org/wiki/Token_bucket

[3] juju/ratelimit https://github.com/juju/ratelimit

[4] B 站在微服務(wù)治理中的探索與實(shí)踐 https://www.infoq.cn/article/zRuGHM_SsQ0lk7gWyBgG

[5] 限流器系列(1) -- Leaky Bucket 漏斗桶 https://www.haohongfan.com/post/2020-06-27-leaky-bucket/

推薦閱讀

• Go 限流器實(shí)戰(zhàn)系列(1) -- Leaky Bucket 漏斗桶

---

喜歡本文的朋友，歡迎關(guān)注“Go語言中文網(wǎng)”：

Go語言中文網(wǎng)啟用微信學(xué)習(xí)交流群，歡迎加微信：274768166，投稿亦歡迎

總結(jié)

以上是生活随笔為你收集整理的令牌桶 限速_Go 限流器实战系列(2) Token Bucket 令牌桶的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。