快捷導航

Golang實現(xiàn)常見的限流算法的示例代碼

更新時間：2023年04月02日 15:07:52 作者：jxwu

限流是項目中經(jīng)常需要使用到的一種工具，一般用于限制用戶的請求的頻率，也可以避免瞬間流量過大導致系統(tǒng)崩潰，或者穩(wěn)定消息處理速率，本文主要介紹了使用Go實現(xiàn)常見的限流算法，希望對大家有所幫助

固定窗口

每開啟一個新的窗口，在窗口時間大小內(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()
   // 獲取當前時間
   now := time.Now()
   // 如果當前窗口失效，計數(shù)器清0，開啟新的窗口
   if now.Sub(l.lastTime) > l.window {
      l.counter = 0
      l.lastTime = now
   }
   // 若到達窗口請求上限，請求失敗
   if l.counter >= l.limit {
      return false
   }
   // 若沒到窗口請求上限，計數(shù)器+1，請求成功
   l.counter++
   return true
}

滑動窗口

滑動窗口類似于固定窗口，它只是把大窗口切分成多個小窗口，每次向右移動一個小窗口，它可以避免兩倍的突發(fā)流量。

固定窗口可以說是滑動窗口的一種特殊情況，只要滑動窗口里面的小窗口和大窗口大小一樣。

窗口算法都有一個問題，當流量達到上限，后面的請求都會被拒絕。

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()

   // 獲取當前小窗口值
   currentSmallWindow := time.Now().UnixNano() / l.smallWindow * l.smallWindow
   // 獲取起始小窗口值
   startSmallWindow := currentSmallWindow - l.smallWindow*(l.smallWindows-1)

   // 計算當前窗口的請求總數(shù)
   var count int
   for smallWindow, counter := range l.counters {
      if smallWindow < startSmallWindow {
         delete(l.counters, smallWindow)
      } else {
         count += counter
      }
   }

   // 若到達窗口請求上限，請求失敗
   if count >= l.limit {
      return false
   }
   // 若沒到窗口請求上限，當前小窗口計數(shù)器+1，請求成功
   l.counters[currentSmallWindow]++
   return true
}

漏桶算法

漏桶是模擬一個漏水的桶，請求相當于往桶里倒水，處理請求的速度相當于水漏出的速度。

主要用于請求處理速率較為穩(wěn)定的服務(wù)，需要使用生產(chǎn)者消費者模式把請求放到一個隊列里，讓消費者以一個較為穩(wěn)定的速率處理。

package limiter

import (
   "sync"
   "time"
)

// LeakyBucketLimiter 漏桶限流器
type LeakyBucketLimiter struct {
   peakLevel       int        // 最高水位
   currentLevel    int        // 當前水位
   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 {
      // 當前水位-距離上次放水的時間(秒)*水流速度
      l.currentLevel = maxInt(0, l.currentLevel-int(interval/time.Second)*l.currentVelocity)
      l.lastTime = now
   }

   // 若到達最高水位，請求失敗
   if l.currentLevel >= l.peakLevel {
      return false
   }
   // 若沒有到達最高水位，當前水位+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 {
      // 當前令牌數(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
   }
   // 如果有令牌，當前令牌-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) {
   // 復制策略避免被修改
   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()

   // 獲取當前小窗口值
   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)
   }

   // 計算每個策略當前窗口的請求總數(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
         }
      }
   }

   // 若到達對應(yīng)策略窗口請求上限，請求失敗，返回違背的策略
   for i, strategy := range l.strategies {
      if counts[i] >= strategy.limit {
         return &ViolationStrategyError{
            Limit:  strategy.limit,
            Window: time.Duration(strategy.window),
         }
      }
   }

   // 若沒到窗口請求上限，當前小窗口計數(shù)器+1，請求成功
   l.counters[currentSmallWindow]++
   return nil
}