您好,登錄后才能下訂單哦!
本篇內(nèi)容介紹了“Golang怎么實現(xiàn)常見的限流算法”的有關(guān)知識,在實際案例的操作過程中,不少人都會遇到這樣的困境,接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧!希望大家仔細(xì)閱讀,能夠?qū)W有所成!
每開啟一個新的窗口,在窗口時間大小內(nèi),可以通過窗口請求上限個請求。
該算法主要是會存在臨界問題,如果流量都集中在兩個窗口的交界處,那么突發(fā)流量會是設(shè)置上限的兩倍。
package limiter import ( "sync" "time" ) // FixedWindowLimiter 固定窗口限流器 type FixedWindowLimiter struct { limit int // 窗口請求上限 window time.Duration // 窗口時間大小 counter int // 計數(shù)器 lastTime time.Time // 上一次請求的時間 mutex sync.Mutex // 避免并發(fā)問題 } 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)前時間 now := time.Now() // 如果當(dāng)前窗口失效,計數(shù)器清0,開啟新的窗口 if now.Sub(l.lastTime) > l.window { l.counter = 0 l.lastTime = now } // 若到達(dá)窗口請求上限,請求失敗 if l.counter >= l.limit { return false } // 若沒到窗口請求上限,計數(shù)器+1,請求成功 l.counter++ return true }
滑動窗口類似于固定窗口,它只是把大窗口切分成多個小窗口,每次向右移動一個小窗口,它可以避免兩倍的突發(fā)流量。
固定窗口可以說是滑動窗口的一種特殊情況,只要滑動窗口里面的小窗口和大窗口大小一樣。
窗口算法都有一個問題,當(dāng)流量達(dá)到上限,后面的請求都會被拒絕。
package limiter import ( "errors" "sync" "time" ) // SlidingWindowLimiter 滑動窗口限流器 type SlidingWindowLimiter struct { limit int // 窗口請求上限 window int64 // 窗口時間大小 smallWindow int64 // 小窗口時間大小 smallWindows int64 // 小窗口數(shù)量 counters map[int64]int // 小窗口計數(shù)器 mutex sync.Mutex // 避免并發(fā)問題 } // NewSlidingWindowLimiter 創(chuàng)建滑動窗口限流器 func NewSlidingWindowLimiter(limit int, window, smallWindow time.Duration) (*SlidingWindowLimiter, error) { // 窗口時間必須能夠被小窗口時間整除 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) // 計算當(dāng)前窗口的請求總數(shù) var count int for smallWindow, counter := range l.counters { if smallWindow < startSmallWindow { delete(l.counters, smallWindow) } else { count += counter } } // 若到達(dá)窗口請求上限,請求失敗 if count >= l.limit { return false } // 若沒到窗口請求上限,當(dāng)前小窗口計數(shù)器+1,請求成功 l.counters[currentSmallWindow]++ return true }
漏桶是模擬一個漏水的桶,請求相當(dāng)于往桶里倒水,處理請求的速度相當(dāng)于水漏出的速度。
主要用于請求處理速率較為穩(wěn)定的服務(wù),需要使用生產(chǎn)者消費者模式把請求放到一個隊列里,讓消費者以一個較為穩(wěn)定的速率處理。
package limiter import ( "sync" "time" ) // LeakyBucketLimiter 漏桶限流器 type LeakyBucketLimiter struct { peakLevel int // 最高水位 currentLevel int // 當(dāng)前水位 currentVelocity int // 水流速度/秒 lastTime time.Time // 上次放水時間 mutex sync.Mutex // 避免并發(fā)問題 } 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() // 距離上次放水的時間 interval := now.Sub(l.lastTime) if interval >= time.Second { // 當(dāng)前水位-距離上次放水的時間(秒)*水流速度 l.currentLevel = maxInt(0, l.currentLevel-int(interval/time.Second)*l.currentVelocity) l.lastTime = now } // 若到達(dá)最高水位,請求失敗 if l.currentLevel >= l.peakLevel { return false } // 若沒有到達(dá)最高水位,當(dāng)前水位+1,請求成功 l.currentLevel++ return true } func maxInt(a, b int) int { if a > b { return a } return b }
與漏桶算法的相反,令牌桶會不斷地把令牌添加到桶里,而請求會從桶中獲取令牌,只有擁有令牌地請求才能被接受。
因為桶中可以提前保留一些令牌,所以它允許一定地突發(fā)流量通過。
package limiter import ( "sync" "time" ) // TokenBucketLimiter 令牌桶限流器 type TokenBucketLimiter struct { capacity int // 容量 currentTokens int // 令牌數(shù)量 rate int // 發(fā)放令牌速率/秒 lastTime time.Time // 上次發(fā)放令牌時間 mutex sync.Mutex // 避免并發(fā)問題 } 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ā)放令牌的時間 interval := now.Sub(l.lastTime) if interval >= time.Second { // 當(dāng)前令牌數(shù)量+距離上次發(fā)放令牌的時間(秒)*發(fā)放令牌速率 l.currentTokens = minInt(l.capacity, l.currentTokens+int(interval/time.Second)*l.rate) l.lastTime = now } // 如果沒有令牌,請求失敗 if l.currentTokens == 0 { return false } // 如果有令牌,當(dāng)前令牌-1,請求成功 l.currentTokens-- return true } func minInt(a, b int) int { if a < b { return a } return b }
滑動日志與滑動窗口算法類似,但是滑動日志主要用于多級限流的場景,比如短信驗證碼1分鐘1次,1小時10次,1天20次這種業(yè)務(wù)。
算法流程與滑動窗口相同,只是它可以指定多個策略,同時在請求失敗的時候,需要通知調(diào)用方是被哪個策略所攔截。
package limiter import ( "errors" "fmt" "sort" "sync" "time" ) // ViolationStrategyError 違背策略錯誤 type ViolationStrategyError struct { Limit int // 窗口請求上限 Window time.Duration // 窗口時間大小 } func (e *ViolationStrategyError) Error() string { return fmt.Sprintf("violation strategy that limit = %d and window = %d", e.Limit, e.Window) } // SlidingLogLimiterStrategy 滑動日志限流器的策略 type SlidingLogLimiterStrategy struct { limit int // 窗口請求上限 window int64 // 窗口時間大小 smallWindows int64 // 小窗口數(shù)量 } func NewSlidingLogLimiterStrategy(limit int, window time.Duration) *SlidingLogLimiterStrategy { return &SlidingLogLimiterStrategy{ limit: limit, window: int64(window), } } // SlidingLogLimiter 滑動日志限流器 type SlidingLogLimiter struct { strategies []*SlidingLogLimiterStrategy // 滑動日志限流器策略列表 smallWindow int64 // 小窗口時間大小 counters map[int64]int // 小窗口計數(shù)器 mutex sync.Mutex // 避免并發(fā)問題 } 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") } // 排序策略,窗口時間大的排前面,相同窗口上限大的排前面 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 { // 隨著窗口時間變小,窗口上限也應(yīng)該變小 if i > 0 { if strategy.limit >= strategies[i-1].limit { return nil, errors.New("the smaller window should be the smaller limit") } } // 窗口時間必須能夠被小窗口時間整除 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 // 獲取每個策略的起始小窗口值 startSmallWindows := make([]int64, len(l.strategies)) for i, strategy := range l.strategies { startSmallWindows[i] = currentSmallWindow - l.smallWindow*(strategy.smallWindows-1) } // 計算每個策略當(dā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á)對應(yīng)策略窗口請求上限,請求失敗,返回違背的策略 for i, strategy := range l.strategies { if counts[i] >= strategy.limit { return &ViolationStrategyError{ Limit: strategy.limit, Window: time.Duration(strategy.window), } } } // 若沒到窗口請求上限,當(dāng)前小窗口計數(shù)器+1,請求成功 l.counters[currentSmallWindow]++ return nil }
“Golang怎么實現(xiàn)常見的限流算法”的內(nèi)容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站,小編將為大家輸出更多高質(zhì)量的實用文章!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。