這篇文章主要介紹了go語(yǔ)言中的限流漏桶和令牌桶庫(kù)怎么使用的相關(guān)知識(shí)，內(nèi)容詳細(xì)易懂，操作簡(jiǎn)單快捷，具有一定借鑒價(jià)值，相信大家閱讀完這篇go語(yǔ)言中的限流漏桶和令牌桶庫(kù)怎么使用文章都會(huì)有所收獲，下面我們一起來(lái)看看吧。
隆陽(yáng)ssl適用于網(wǎng)站、小程序/APP、API接口等需要進(jìn)行數(shù)據(jù)傳輸應(yīng)用場(chǎng)景，ssl證書(shū)未來(lái)市場(chǎng)廣闊！成為創(chuàng)新互聯(lián)建站的ssl證書(shū)銷(xiāo)售渠道，可以享受市場(chǎng)價(jià)格4-6折優(yōu)惠！如果有意向歡迎電話聯(lián)系或者加微信：13518219792（備注：SSL證書(shū)合作）期待與您的合作！
在大數(shù)據(jù)量高并發(fā)訪問(wèn)時(shí)，經(jīng)常會(huì)出現(xiàn)服務(wù)或接口面對(duì)大量的請(qǐng)求而導(dǎo)致數(shù)據(jù)庫(kù)崩潰的情況，甚至引發(fā)連鎖反映導(dǎo)致整個(gè)系統(tǒng)崩潰?；蛘哂腥藧阂夤艟W(wǎng)站，大量的無(wú)用請(qǐng)求出現(xiàn)會(huì)導(dǎo)致緩存穿透的情況出現(xiàn)。使用限流中間件可以在短時(shí)間內(nèi)對(duì)請(qǐng)求進(jìn)行限制數(shù)量，起到降級(jí)的作用，從而保障了網(wǎng)站的安全性。
使用消息中間件進(jìn)行統(tǒng)一限制（降速）
使用限流方案將多余請(qǐng)求返回（限流）
升級(jí)服務(wù)器
緩存（但仍然有緩存穿透等危險(xiǎn)）
等等
可以看出在代碼已經(jīng)無(wú)法提升的情況下，只能去提升硬件水平?；蛘吒膭?dòng)架構(gòu)再加一層！也可以使用消息中間件統(tǒng)一處理。而結(jié)合看來(lái)，限流方案是一種既不需要大幅改動(dòng)也不需要高額開(kāi)銷(xiāo)的策略。
令牌桶算法
漏桶算法
滑動(dòng)窗口算法
等等
go get -u go.uber.org/ratelimit
// New returns a Limiter that will limit to the given RPS.
func New(rate int, opts ...Option) Limiter {
return newAtomicBased(rate, opts...)
}
// newAtomicBased returns a new atomic based limiter.
func newAtomicBased(rate int, opts ...Option) *atomicLimiter {
// TODO consider moving config building to the implementation
// independent code.
config := buildConfig(opts)
perRequest := config.per / time.Duration(rate)
l := &atomicLimiter{
perRequest: perRequest,
maxSlack: -1 * time.Duration(config.slack) * perRequest,
clock: config.clock,
}
initialState := state{
last: time.Time{},
sleepFor: 0,
}
atomic.StorePointer(&l.state, unsafe.Pointer(&initialState))
return l
}
該函數(shù)使用了函數(shù)選項(xiàng)模式對(duì)多個(gè)結(jié)構(gòu)體對(duì)象進(jìn)行初始化
根據(jù)傳入的值來(lái)初始化一個(gè)桶結(jié)構(gòu)體 rate 為int 傳參 。
初始化過(guò)程中包括了
每一滴水需要的時(shí)間 perquest = config.per / time.Duration(rate)
maxSlack 寬松度（寬松度為負(fù)值）-1 * time.Duration(config.slack) * perRequest 松緊度是用來(lái)規(guī)范等待時(shí)間的
// Clock is the minimum necessary interface to instantiate a rate limiter with
// a clock or mock clock, compatible with clocks created using
// github.com/andres-erbsen/clock.
type Clock interface {
Now() time.Time
Sleep(time.Duration)
}
同時(shí)還需要結(jié)構(gòu)體Clock來(lái)記錄當(dāng)前請(qǐng)求的時(shí)間now和此刻的請(qǐng)求所需要花費(fèi)等待的時(shí)間sleep
type state struct {
last time.Time
sleepFor time.Duration
}
state 主要用來(lái)記錄上次執(zhí)行的時(shí)間以及當(dāng)前執(zhí)行請(qǐng)求需要花費(fèi)等待的時(shí)間（作為中間狀態(tài)記錄）
func (t *atomicLimiter) Take() time.Time {
var (
newState state
taken bool
interval time.Duration
)
for !taken {
now := t.clock.Now()
previousStatePointer := atomic.LoadPointer(&t.state)
oldState := (*state)(previousStatePointer)
newState = state{
last: now,
sleepFor: oldState.sleepFor,
}
if oldState.last.IsZero() {
taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
continue
}
// 計(jì)算是否需要進(jìn)行等待取水操作
newState.sleepFor += t.perRequest(每?jī)傻嗡g的間隔時(shí)間) - now.Sub(oldState.last)(當(dāng)前時(shí)間與上次取水時(shí)間的間隔)
// 如果等待取水時(shí)間特別小，就需要松緊度進(jìn)行維護(hù)
if newState.sleepFor < t.maxSlack {
newState.sleepFor = t.maxSlack
}
// 如果等待時(shí)間大于0，就進(jìn)行更新
if newState.sleepFor > 0 {
newState.last = newState.last.Add(newState.sleepFor)
interval, newState.sleepFor = newState.sleepFor, 0
}
taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
}
t.clock.Sleep(interval)
// 最后返回需要等待的時(shí)間
return newState.last
}
實(shí)現(xiàn)一個(gè)Take方法
該Take方法會(huì)進(jìn)行原子性操作(可以理解為加鎖和解鎖)，在大量并發(fā)請(qǐng)求下仍可以保證正常使用。
記錄下當(dāng)前的時(shí)間 now := t.clock.Now()
oldState.last.IsZero()判斷是不是第一次取水，如果是就直接將state結(jié)構(gòu)體中的值進(jìn)行返回。而這個(gè)結(jié)構(gòu)體中初始化了上次執(zhí)行時(shí)間，如果是第一次取水就作為當(dāng)前時(shí)間直接傳參。
如果 newState.sleepFor 非常小，就會(huì)出現(xiàn)問(wèn)題，因此需要借助寬松度，一旦這個(gè)最小值比寬松度小，就用寬松度對(duì)取水時(shí)間進(jìn)行維護(hù)。
如果newState.sleepFor > 0 就直接更新結(jié)構(gòu)體中上次執(zhí)行時(shí)間newState.last = newState.last.Add(newState.sleepFor)并記錄需要等待的時(shí)間interval, newState.sleepFor = newState.sleepFor, 0。
如果允許取水和等待操作，那就說(shuō)明沒(méi)有發(fā)生并發(fā)競(jìng)爭(zhēng)的情況，就模擬睡眠時(shí)間t.clock.Sleep(interval)。然后將取水的目標(biāo)時(shí)間進(jìn)行返回，由服務(wù)端代碼來(lái)判斷是否打回響應(yīng)或者等待該時(shí)間后繼續(xù)響應(yīng)。
func (c *clock) Sleep(d time.Duration) { time.Sleep(d) }
實(shí)際上在一個(gè)請(qǐng)求來(lái)的時(shí)候，限流器就會(huì)進(jìn)行睡眠對(duì)應(yīng)的時(shí)間，并在睡眠后將最新取水時(shí)間返回。
func ratelimit1() func(ctx *gin.Context) {
r1 := rate1.New(100)
return func(ctx *gin.Context) {
now := time.Now()
// Take 返回的是一個(gè) time.Duration的時(shí)間
if r1.Take().Sub(now) > 0 {
// 返回的時(shí)間比當(dāng)前的時(shí)間還大，說(shuō)明需要進(jìn)行等待
// 如果需要等待, 就 time.Sleep(r1.Take().Sub(now())) 然后放行
// 如果不需要等待請(qǐng)求時(shí)間，就直接進(jìn)行Abort 然后返回
response(ctx, http.StatusRequestTimeout, "rate1 limit...")
fmt.Println("rate1 limit...")
ctx.Abort()
return
}
// 放行
ctx.Next()
}
}
這里你可以進(jìn)行選擇是否返回。因?yàn)門(mén)ake一定會(huì)執(zhí)行sleep函數(shù)，所以當(dāng)執(zhí)行take結(jié)束后表示當(dāng)前請(qǐng)求已經(jīng)接到了水。當(dāng)前演示使用第一種情況。
如果你的業(yè)務(wù)要求響應(yīng)不允許進(jìn)行等待。那么可以在該請(qǐng)求接完水之后然后，如上例。
如果你的業(yè)務(wù)允許響應(yīng)等待，那么該請(qǐng)求等待對(duì)應(yīng)的接水時(shí)間后進(jìn)行下一步。具體代碼就是將if中的內(nèi)容直接忽略。（建議使用）
這里定義了一個(gè)響應(yīng)函數(shù)和一個(gè)handler函數(shù)方便測(cè)試
func response(c *gin.Context, code int, info any) {
c.JSON(code, info)
}
func pingHandler(c *gin.Context) {
response(c, 200, "ping ok~")
}
執(zhí)行go test -run=Run -v先開(kāi)啟一個(gè)web服務(wù)
func TestRun(t *testing.T) {
r := gin.Default()
r.GET("/ping1", ratelimit1(), pingHandler)
r.GET("/ping2", ratelimit2(), helloHandler)
_ = r.Run(":4399")
}
使用接口壓力測(cè)試工具go-wrk進(jìn)行測(cè)試->tsliwowicz/go-wrk: go-wrk)
在golang引入install版本可以直接通過(guò)go install github.com/tsliwowicz/go-wrk@latest下載
Usage: go-wrk
-t 8個(gè)線程 -c 400個(gè)連接 -n 模擬100次請(qǐng)求 -d 替換-n 表示連接時(shí)間
輸入
go-wrk -t=8 -c=400 -n=100 http://127.0.0.1:4399/ping1
可以稍微等待一下水流積攢（壓測(cè)速度過(guò)快）。
可以看出，89個(gè)請(qǐng)求全部返回。也就是說(shuō)在一段請(qǐng)求高峰期，不會(huì)有請(qǐng)求進(jìn)行響應(yīng)。因此我認(rèn)為既然內(nèi)部已經(jīng)睡眠，那么就也就應(yīng)該對(duì)請(qǐng)求放行處理。
引入ratelimit庫(kù)
go get -u github.com/juju/ratelimit
// NewBucket returns a new token bucket that fills at the
// rate of one token every fillInterval, up to the given
// maximum capacity. Both arguments must be
// positive. The bucket is initially full.
func NewBucket(fillInterval time.Duration, capacity int64) *Bucket {
return NewBucketWithClock(fillInterval, capacity, nil)
}
// NewBucketWithClock is identical to NewBucket but injects a testable clock
// interface.
func NewBucketWithClock(fillInterval time.Duration, capacity int64, clock Clock) *Bucket {
return NewBucketWithQuantumAndClock(fillInterval, capacity, 1, clock)
}
進(jìn)行Bucket桶的初始化。
func NewBucketWithQuantumAndClock(fillInterval time.Duration, capacity, quantum int64, clock Clock) *Bucket {
if clock == nil {
clock = realClock{}
}
// 填充速率
if fillInterval <= 0 {
panic("token bucket fill interval is not > 0")
}
// 最大令牌容量
if capacity <= 0 {
panic("token bucket capacity is not > 0")
}
// 單次令牌生成量
if quantum <= 0 {
panic("token bucket quantum is not > 0")
}
return &Bucket{
clock: clock,
startTime: clock.Now(),
latestTick: 0,
fillInterval: fillInterval,
capacity: capacity,
quantum: quantum,
availableTokens: capacity,
}
}
令牌桶初始化過(guò)程，初始化結(jié)構(gòu)體 fillInterval(填充速率) cap(最大令牌量) quannum(每次令牌生成量)。
如果三個(gè)變量有一個(gè)小于或者等于0的話直接進(jìn)行報(bào)錯(cuò)返回。在最開(kāi)始就將當(dāng)前令牌數(shù)初始化為最大容量。
// TakeAvailable takes up to count immediately available tokens from the
// bucket. It returns the number of tokens removed, or zero if there are
// no available tokens. It does not block.
func (tb *Bucket) TakeAvailable(count int64) int64 {
tb.mu.Lock()
defer tb.mu.Unlock()
return tb.takeAvailable(tb.clock.Now(), count)
}
調(diào)用TakeAvailable函數(shù)，傳入?yún)?shù)為需要取出的令牌數(shù)量，返回參數(shù)是實(shí)際能夠取出的令牌數(shù)量。
func (tb *Bucket) takeAvailable(now time.Time, count int64) int64 {
// 如果需要取出的令牌數(shù)小于等于零，那么就返回0個(gè)令牌
if count <= 0 {
return 0
}
// 根據(jù)時(shí)間對(duì)當(dāng)前桶中令牌數(shù)進(jìn)行計(jì)算
tb.adjustavailableTokens(tb.currentTick(now))
// 計(jì)算之后的令牌總數(shù)小于等于0，說(shuō)明當(dāng)前令牌不足取出，那么就直接返回0個(gè)令牌
if tb.availableTokens <= 0 {
return 0
}
// 如果當(dāng)前存儲(chǔ)的令牌數(shù)量多于請(qǐng)求數(shù)量，那么就返回取出令牌數(shù)
if count > tb.availableTokens {
count = tb.availableTokens
}
// 調(diào)整令牌數(shù)
tb.availableTokens -= count
return count
}
如果需要取出的令牌數(shù)小于等于零，那么就返回0個(gè)令牌
根據(jù)時(shí)間對(duì)當(dāng)前桶中令牌數(shù)進(jìn)行計(jì)算
計(jì)算之后的令牌總數(shù)小于等于0，說(shuō)明當(dāng)前令牌不足取出，那么就直接返回0個(gè)令牌
如果當(dāng)前存儲(chǔ)的令牌數(shù)量多于請(qǐng)求數(shù)量，那么就返回取出令牌數(shù)
調(diào)整令牌數(shù)
func (tb *Bucket) adjustavailableTokens(tick int64) {
lastTick := tb.latestTick
tb.latestTick = tick
// 如果當(dāng)前令牌數(shù)大于最大等于容量，直接返回最大容量
if tb.availableTokens >= tb.capacity {
return
}
// 當(dāng)前令牌數(shù) += (當(dāng)前時(shí)間 - 上次取出令牌數(shù)的時(shí)間) * quannum(每次生成令牌量)
tb.availableTokens += (tick - lastTick) * tb.quantum
// 如果當(dāng)前令牌數(shù)大于最大等于容量, 將當(dāng)前令牌數(shù) = 最大容量 然后返回 當(dāng)前令牌數(shù)
if tb.availableTokens > tb.capacity {
tb.availableTokens = tb.capacity
}
return
}
如果當(dāng)前令牌數(shù)大于最大等于容量，直接返回最大容量
當(dāng)前令牌數(shù) += (當(dāng)前時(shí)間 - 上次取出令牌數(shù)的時(shí)間) * quannum(每次生成令牌量)
如果當(dāng)前令牌數(shù)大于最大等于容量, 將當(dāng)前令牌數(shù) = 最大容量 然后返回 當(dāng)前令牌數(shù)
加鎖 defer 解鎖
判斷count(想要取出的令牌數(shù)) 是否小于等于 0，如果是直接返回 0
調(diào)用函數(shù)adjustTokens 獲取可用的令牌數(shù)量
如果當(dāng)前可以取出的令牌數(shù)小于等于0 直接返回 0
如果當(dāng)前可以取出的令牌數(shù)小于當(dāng)前想要取出的令牌數(shù)(count) count = 當(dāng)前可以取出的令牌數(shù)
當(dāng)前的令牌數(shù) -= 取出的令牌數(shù) (count)
返回 count(可以取出的令牌數(shù))
take函數(shù)，能夠返回等待時(shí)間和布爾值，允許欠賬，沒(méi)有令牌也可以取出。
func (tb *Bucket) Take(count int64) time.Duration
takeMaxDuration函數(shù)，可以根據(jù)最大等待時(shí)間來(lái)進(jìn)行判斷。
func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool)
因?yàn)樗麄儍?nèi)部的實(shí)現(xiàn)都基于令牌調(diào)整，我這里不做過(guò)多介紹，如果感興趣可以自行研究一下。
func ratelimit2() func(ctx *gin.Context) {
// 生成速率 最大容量
r2 := rate2.NewBucket(time.Second, 200)
return func(ctx *gin.Context) {
//r2.Take() // 允許欠賬，令牌不夠也可以接收請(qǐng)求
if r2.TakeAvailable(1) == 1 {
// 如果想要取出1個(gè)令牌并且能夠取出，就放行
ctx.Next()
return
}
response(ctx, http.StatusRequestTimeout, "rate2 limit...")
ctx.Abort()
return
}
}
壓測(cè)速度過(guò)于快速，在實(shí)際過(guò)程中可以根據(jù)調(diào)整令牌生成速率來(lái)進(jìn)行具體限流！
關(guān)于“go語(yǔ)言中的限流漏桶和令牌桶庫(kù)怎么使用”這篇文章的內(nèi)容就介紹到這里，感謝各位的閱讀！相信大家對(duì)“go語(yǔ)言中的限流漏桶和令牌桶庫(kù)怎么使用”知識(shí)都有一定的了解，大家如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。
重慶分公司
重慶分公司
