Golang實(shí)現(xiàn)常見(jiàn)的限流算法的示例代碼
限流是項(xiàng)目中經(jīng)常需要使用到的一種工具,一般用于限制用戶的請(qǐng)求的頻率,也可以避免瞬間流量過(guò)大導(dǎo)致系統(tǒng)崩潰,或者穩(wěn)定消息處理速率
這個(gè)文章主要是使用Go實(shí)現(xiàn)常見(jiàn)的限流算法,代碼參考了文章面試官:來(lái),年輕人!請(qǐng)手?jǐn)]5種常見(jiàn)限流算法! 和面試必備:4種經(jīng)典限流算法講解如果需要Java實(shí)現(xiàn)或更詳細(xì)的算法介紹可以看這兩篇文章
固定窗口
每開(kāi)啟一個(gè)新的窗口,在窗口時(shí)間大小內(nèi),可以通過(guò)窗口請(qǐng)求上限個(gè)請(qǐng)求。
該算法主要是會(huì)存在臨界問(wèn)題,如果流量都集中在兩個(gè)窗口的交界處,那么突發(fā)流量會(huì)是設(shè)置上限的兩倍。
package limiter import ( "sync" "time" ) // FixedWindowLimiter 固定窗口限流器 type FixedWindowLimiter struct { limit int // 窗口請(qǐng)求上限 window time.Duration // 窗口時(shí)間大小 counter int // 計(jì)數(shù)器 lastTime time.Time // 上一次請(qǐng)求的時(shí)間 mutex sync.Mutex // 避免并發(fā)問(wèn)題 } func NewFixedWindowLimiter(limit int, window time.Duration) *FixedWindowLimiter { return &FixedWindowLimiter{ limit: limit, window: window, lastTime: time.Now(), } } func (l *FixedWindowLimiter) TryAcquire() bool { l.mutex.Lock() defer l.mutex.Unlock() // 獲取當(dāng)前時(shí)間 now := time.Now() // 如果當(dāng)前窗口失效,計(jì)數(shù)器清0,開(kāi)啟新的窗口 if now.Sub(l.lastTime) > l.window { l.counter = 0 l.lastTime = now } // 若到達(dá)窗口請(qǐng)求上限,請(qǐng)求失敗 if l.counter >= l.limit { return false } // 若沒(méi)到窗口請(qǐng)求上限,計(jì)數(shù)器+1,請(qǐng)求成功 l.counter++ return true }
滑動(dòng)窗口
滑動(dòng)窗口類似于固定窗口,它只是把大窗口切分成多個(gè)小窗口,每次向右移動(dòng)一個(gè)小窗口,它可以避免兩倍的突發(fā)流量。
固定窗口可以說(shuō)是滑動(dòng)窗口的一種特殊情況,只要滑動(dòng)窗口里面的小窗口和大窗口大小一樣。
窗口算法都有一個(gè)問(wèn)題,當(dāng)流量達(dá)到上限,后面的請(qǐng)求都會(huì)被拒絕。
package limiter import ( "errors" "sync" "time" ) // SlidingWindowLimiter 滑動(dòng)窗口限流器 type SlidingWindowLimiter struct { limit int // 窗口請(qǐng)求上限 window int64 // 窗口時(shí)間大小 smallWindow int64 // 小窗口時(shí)間大小 smallWindows int64 // 小窗口數(shù)量 counters map[int64]int // 小窗口計(jì)數(shù)器 mutex sync.Mutex // 避免并發(fā)問(wèn)題 } // NewSlidingWindowLimiter 創(chuàng)建滑動(dòng)窗口限流器 func NewSlidingWindowLimiter(limit int, window, smallWindow time.Duration) (*SlidingWindowLimiter, error) { // 窗口時(shí)間必須能夠被小窗口時(shí)間整除 if window%smallWindow != 0 { return nil, errors.New("window cannot be split by integers") } return &SlidingWindowLimiter{ limit: limit, window: int64(window), smallWindow: int64(smallWindow), smallWindows: int64(window / smallWindow), counters: make(map[int64]int), }, nil } func (l *SlidingWindowLimiter) TryAcquire() bool { l.mutex.Lock() defer l.mutex.Unlock() // 獲取當(dāng)前小窗口值 currentSmallWindow := time.Now().UnixNano() / l.smallWindow * l.smallWindow // 獲取起始小窗口值 startSmallWindow := currentSmallWindow - l.smallWindow*(l.smallWindows-1) // 計(jì)算當(dāng)前窗口的請(qǐng)求總數(shù) var count int for smallWindow, counter := range l.counters { if smallWindow < startSmallWindow { delete(l.counters, smallWindow) } else { count += counter } } // 若到達(dá)窗口請(qǐng)求上限,請(qǐng)求失敗 if count >= l.limit { return false } // 若沒(méi)到窗口請(qǐng)求上限,當(dāng)前小窗口計(jì)數(shù)器+1,請(qǐng)求成功 l.counters[currentSmallWindow]++ return true }
漏桶算法
漏桶是模擬一個(gè)漏水的桶,請(qǐng)求相當(dāng)于往桶里倒水,處理請(qǐng)求的速度相當(dāng)于水漏出的速度。
主要用于請(qǐng)求處理速率較為穩(wěn)定的服務(wù),需要使用生產(chǎn)者消費(fèi)者模式把請(qǐng)求放到一個(gè)隊(duì)列里,讓消費(fèi)者以一個(gè)較為穩(wěn)定的速率處理。
package limiter import ( "sync" "time" ) // LeakyBucketLimiter 漏桶限流器 type LeakyBucketLimiter struct { peakLevel int // 最高水位 currentLevel int // 當(dāng)前水位 currentVelocity int // 水流速度/秒 lastTime time.Time // 上次放水時(shí)間 mutex sync.Mutex // 避免并發(fā)問(wèn)題 } func NewLeakyBucketLimiter(peakLevel, currentVelocity int) *LeakyBucketLimiter { return &LeakyBucketLimiter{ peakLevel: peakLevel, currentVelocity: currentVelocity, lastTime: time.Now(), } } func (l *LeakyBucketLimiter) TryAcquire() bool { l.mutex.Lock() defer l.mutex.Unlock() // 嘗試放水 now := time.Now() // 距離上次放水的時(shí)間 interval := now.Sub(l.lastTime) if interval >= time.Second { // 當(dāng)前水位-距離上次放水的時(shí)間(秒)*水流速度 l.currentLevel = maxInt(0, l.currentLevel-int(interval/time.Second)*l.currentVelocity) l.lastTime = now } // 若到達(dá)最高水位,請(qǐng)求失敗 if l.currentLevel >= l.peakLevel { return false } // 若沒(méi)有到達(dá)最高水位,當(dāng)前水位+1,請(qǐng)求成功 l.currentLevel++ return true } func maxInt(a, b int) int { if a > b { return a } return b }
令牌桶
與漏桶算法的相反,令牌桶會(huì)不斷地把令牌添加到桶里,而請(qǐng)求會(huì)從桶中獲取令牌,只有擁有令牌地請(qǐng)求才能被接受。
因?yàn)橥爸锌梢蕴崆氨A粢恍┝钆?,所以它允許一定地突發(fā)流量通過(guò)。
package limiter import ( "sync" "time" ) // TokenBucketLimiter 令牌桶限流器 type TokenBucketLimiter struct { capacity int // 容量 currentTokens int // 令牌數(shù)量 rate int // 發(fā)放令牌速率/秒 lastTime time.Time // 上次發(fā)放令牌時(shí)間 mutex sync.Mutex // 避免并發(fā)問(wèn)題 } func NewTokenBucketLimiter(capacity, rate int) *TokenBucketLimiter { return &TokenBucketLimiter{ capacity: capacity, rate: rate, lastTime: time.Now(), } } func (l *TokenBucketLimiter) TryAcquire() bool { l.mutex.Lock() defer l.mutex.Unlock() // 嘗試發(fā)放令牌 now := time.Now() // 距離上次發(fā)放令牌的時(shí)間 interval := now.Sub(l.lastTime) if interval >= time.Second { // 當(dāng)前令牌數(shù)量+距離上次發(fā)放令牌的時(shí)間(秒)*發(fā)放令牌速率 l.currentTokens = minInt(l.capacity, l.currentTokens+int(interval/time.Second)*l.rate) l.lastTime = now } // 如果沒(méi)有令牌,請(qǐng)求失敗 if l.currentTokens == 0 { return false } // 如果有令牌,當(dāng)前令牌-1,請(qǐng)求成功 l.currentTokens-- return true } func minInt(a, b int) int { if a < b { return a } return b }
滑動(dòng)日志
滑動(dòng)日志與滑動(dòng)窗口算法類似,但是滑動(dòng)日志主要用于多級(jí)限流的場(chǎng)景,比如短信驗(yàn)證碼1分鐘1次,1小時(shí)10次,1天20次這種業(yè)務(wù)。
算法流程與滑動(dòng)窗口相同,只是它可以指定多個(gè)策略,同時(shí)在請(qǐng)求失敗的時(shí)候,需要通知調(diào)用方是被哪個(gè)策略所攔截。
package limiter import ( "errors" "fmt" "sort" "sync" "time" ) // ViolationStrategyError 違背策略錯(cuò)誤 type ViolationStrategyError struct { Limit int // 窗口請(qǐng)求上限 Window time.Duration // 窗口時(shí)間大小 } func (e *ViolationStrategyError) Error() string { return fmt.Sprintf("violation strategy that limit = %d and window = %d", e.Limit, e.Window) } // SlidingLogLimiterStrategy 滑動(dòng)日志限流器的策略 type SlidingLogLimiterStrategy struct { limit int // 窗口請(qǐng)求上限 window int64 // 窗口時(shí)間大小 smallWindows int64 // 小窗口數(shù)量 } func NewSlidingLogLimiterStrategy(limit int, window time.Duration) *SlidingLogLimiterStrategy { return &SlidingLogLimiterStrategy{ limit: limit, window: int64(window), } } // SlidingLogLimiter 滑動(dòng)日志限流器 type SlidingLogLimiter struct { strategies []*SlidingLogLimiterStrategy // 滑動(dòng)日志限流器策略列表 smallWindow int64 // 小窗口時(shí)間大小 counters map[int64]int // 小窗口計(jì)數(shù)器 mutex sync.Mutex // 避免并發(fā)問(wèn)題 } func NewSlidingLogLimiter(smallWindow time.Duration, strategies ...*SlidingLogLimiterStrategy) (*SlidingLogLimiter, error) { // 復(fù)制策略避免被修改 strategies = append(make([]*SlidingLogLimiterStrategy, 0, len(strategies)), strategies...) // 不能不設(shè)置策略 if len(strategies) == 0 { return nil, errors.New("must be set strategies") } // 排序策略,窗口時(shí)間大的排前面,相同窗口上限大的排前面 sort.Slice(strategies, func(i, j int) bool { a, b := strategies[i], strategies[j] if a.window == b.window { return a.limit > b.limit } return a.window > b.window }) fmt.Println(strategies[0], strategies[1]) for i, strategy := range strategies { // 隨著窗口時(shí)間變小,窗口上限也應(yīng)該變小 if i > 0 { if strategy.limit >= strategies[i-1].limit { return nil, errors.New("the smaller window should be the smaller limit") } } // 窗口時(shí)間必須能夠被小窗口時(shí)間整除 if strategy.window%int64(smallWindow) != 0 { return nil, errors.New("window cannot be split by integers") } strategy.smallWindows = strategy.window / int64(smallWindow) } return &SlidingLogLimiter{ strategies: strategies, smallWindow: int64(smallWindow), counters: make(map[int64]int), }, nil } func (l *SlidingLogLimiter) TryAcquire() error { l.mutex.Lock() defer l.mutex.Unlock() // 獲取當(dāng)前小窗口值 currentSmallWindow := time.Now().UnixNano() / l.smallWindow * l.smallWindow // 獲取每個(gè)策略的起始小窗口值 startSmallWindows := make([]int64, len(l.strategies)) for i, strategy := range l.strategies { startSmallWindows[i] = currentSmallWindow - l.smallWindow*(strategy.smallWindows-1) } // 計(jì)算每個(gè)策略當(dāng)前窗口的請(qǐng)求總數(shù) counts := make([]int, len(l.strategies)) for smallWindow, counter := range l.counters { if smallWindow < startSmallWindows[0] { delete(l.counters, smallWindow) continue } for i := range l.strategies { if smallWindow >= startSmallWindows[i] { counts[i] += counter } } } // 若到達(dá)對(duì)應(yīng)策略窗口請(qǐng)求上限,請(qǐng)求失敗,返回違背的策略 for i, strategy := range l.strategies { if counts[i] >= strategy.limit { return &ViolationStrategyError{ Limit: strategy.limit, Window: time.Duration(strategy.window), } } } // 若沒(méi)到窗口請(qǐng)求上限,當(dāng)前小窗口計(jì)數(shù)器+1,請(qǐng)求成功 l.counters[currentSmallWindow]++ return nil }
總結(jié)
- 如果需要一個(gè)簡(jiǎn)單高效的算法,可以使用固定窗口,但是它可能產(chǎn)生兩倍的突發(fā)流量
- 可以通過(guò)滑動(dòng)窗口避免突發(fā)流量問(wèn)題,但是窗口可能會(huì)掐斷流量一段時(shí)間
- 如果需要更平滑的流量,可以使用漏桶算法搭配生產(chǎn)者消費(fèi)者模式
- 如果能夠處理一定的突發(fā)流量,可以使用令牌桶算法
- 遇到多級(jí)限流的場(chǎng)景,滑動(dòng)日志會(huì)更加適合
全部源碼:https://github.com/jiaxwu/limiter
到此這篇關(guān)于Golang實(shí)現(xiàn)常見(jiàn)的限流算法的示例代碼的文章就介紹到這了,更多相關(guān)Golang限流算法內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- golang高并發(fā)限流操作 ping / telnet
- 詳解Golang實(shí)現(xiàn)請(qǐng)求限流的幾種辦法
- golang?熔斷器的實(shí)現(xiàn)過(guò)程
- Golang官方限流器庫(kù)實(shí)現(xiàn)限流示例詳解
- Golang官方限流器time/rate的使用與實(shí)現(xiàn)詳解
- Golang熔斷器的開(kāi)發(fā)過(guò)程詳解
- golang限流庫(kù)兩個(gè)大bug(半年之久無(wú)人提起)
- Golang限流器time/rate設(shè)計(jì)與實(shí)現(xiàn)詳解
- golang 熔斷限流降級(jí)實(shí)踐
相關(guān)文章
Golang中文件目錄操作的實(shí)現(xiàn)步驟詳解
在Golang中,文件目錄是指計(jì)算機(jī)文件系統(tǒng)中的文件夾或目錄。目錄是用于組織和存儲(chǔ)文件的一種方式,可以包含文件和其他子目錄,本文主要介紹了Golang中文件目錄操作的實(shí)現(xiàn)方法,需要的朋友可以參考下2023-05-05Go語(yǔ)言Grpc?Stream的實(shí)現(xiàn)
本文主要介紹了Go語(yǔ)言Grpc?Stream的實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2022-06-06golang 刪除切片的某個(gè)元素及剔除切片內(nèi)的零值方式
這篇文章主要介紹了golang 刪除切片的某個(gè)元素及剔除切片內(nèi)的零值方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-04-04Golang 函數(shù)執(zhí)行時(shí)間統(tǒng)計(jì)裝飾器的一個(gè)實(shí)現(xiàn)詳解
這篇文章主要介紹了Golang 函數(shù)執(zhí)行時(shí)間統(tǒng)計(jì)裝飾器的一個(gè)實(shí)現(xiàn)詳解,小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2019-03-03Go語(yǔ)言中break label與goto label的區(qū)別
這篇文章主要介紹了Go語(yǔ)言中break label與goto label的區(qū)別,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-04-04GoLang協(xié)程庫(kù)libtask學(xué)習(xí)筆記
libtask一個(gè)C語(yǔ)言的協(xié)程庫(kù),是go語(yǔ)言的前身很早期的原型. 測(cè)試機(jī)器是我的mac air 安裝的centos虛擬機(jī)(只有一個(gè)核), 代碼沒(méi)有采用任何優(yōu)化,只是使用默認(rèn)配置2022-12-12