一文詳解限流算法的實(shí)現(xiàn)方式

不依賴外部庫(kù)的情況下，限流算法有什么實(shí)現(xiàn)的思路？本文介紹了3種實(shí)現(xiàn)限流的方式。

一、漏桶算法

• 算法思想 與令牌桶是“反向”的算法，當(dāng)有請(qǐng)求到來(lái)時(shí)先放到木桶中，worker以固定的速度從木桶中取出請(qǐng)求進(jìn)行相應(yīng)。如果木桶已經(jīng)滿了，直接返回請(qǐng)求頻率超限的錯(cuò)誤碼或者頁(yè)面
• 適用場(chǎng)景
  流量最均勻的限流方式，一般用于流量“整形”，例如保護(hù)數(shù)據(jù)庫(kù)的限流。先把對(duì)數(shù)據(jù)庫(kù)的訪問(wèn)加入到木桶中，worker再以db能夠承受的qps從木桶中取出請(qǐng)求，去訪問(wèn)數(shù)據(jù)庫(kù)。不太適合電商搶購(gòu)和微博出現(xiàn)熱點(diǎn)事件等場(chǎng)景的限流，一是應(yīng)對(duì)突發(fā)流量不是很靈活，二是為每個(gè)user_id/ip維護(hù)一個(gè)隊(duì)列(木桶)，workder從這些隊(duì)列中拉取任務(wù)，資源的消耗會(huì)比較大。

• go語(yǔ)言實(shí)現(xiàn)
  通常使用隊(duì)列來(lái)實(shí)現(xiàn)，在go語(yǔ)言中可以通過(guò)buffered channel來(lái)快速實(shí)現(xiàn)，任務(wù)加入channel，開(kāi)啟一定數(shù)量的worker從channel中獲取任務(wù)執(zhí)行。

package main
import ( "fmt" "sync" "time")
// 每個(gè)請(qǐng)求來(lái)了，把需要執(zhí)行的業(yè)務(wù)邏輯封裝成Task，放入木桶，等待worker取出執(zhí)行type Task struct { handler func() Result // worker從木桶中取出請(qǐng)求對(duì)象后要執(zhí)行的業(yè)務(wù)邏輯函數(shù) resChan chan Result   // 等待worker執(zhí)行并返回結(jié)果的channel taskID  int}
// 封裝業(yè)務(wù)邏輯的執(zhí)行結(jié)果type Result struct {}
// 模擬業(yè)務(wù)邏輯的函數(shù)func handler() Result { time.Sleep(300 * time.Millisecond) return Result{}}
func NewTask(id int) Task { return Task{  handler: handler,  resChan: make(chan Result),  taskID:  id, }}
// 漏桶type LeakyBucket struct { BucketSize int       // 木桶的大小 NumWorker  int       // 同時(shí)從木桶中獲取任務(wù)執(zhí)行的worker數(shù)量 bucket     chan Task // 存方任務(wù)的木桶}
func NewLeakyBucket(bucketSize int, numWorker int) *LeakyBucket { return &LeakyBucket{  BucketSize: bucketSize,  NumWorker:  numWorker,  bucket:     make(chan Task, bucketSize), }}
func (b *LeakyBucket) validate(task Task) bool { // 如果木桶已經(jīng)滿了，返回false select { case b.bucket <- task: default:  fmt.Printf("request[id=%d] is refused
", task.taskID)  return false }
 // 等待worker執(zhí)行 <-task.resChan fmt.Printf("request[id=%d] is run
", task.taskID) return true}
func (b *LeakyBucket) Start() { // 開(kāi)啟worker從木桶拉取任務(wù)執(zhí)行 go func() {  for i := 0; i < b.NumWorker; i++ {   go func() {    for {     task := <-b.bucket     result := task.handler()     task.resChan <- result    }   }()  } }()}
func main() { bucket := NewLeakyBucket(10, 4) bucket.Start()
 var wg sync.WaitGroup for i := 0; i < 20; i++ {  wg.Add(1)  go func(id int) {   defer wg.Done()   task := NewTask(id)   bucket.validate(task)  }(i) } wg.Wait()}

二、令牌桶算法

• 算法思想
  想象有一個(gè)木桶，以固定的速度往木桶里加入令牌，木桶滿了則不再加入令牌。服務(wù)收到請(qǐng)求時(shí)嘗試從木桶中取出一個(gè)令牌，如果能夠得到令牌則繼續(xù)執(zhí)行后續(xù)的業(yè)務(wù)邏輯；如果沒(méi)有得到令牌，直接返回反問(wèn)頻率超限的錯(cuò)誤碼或頁(yè)面等，不繼續(xù)執(zhí)行后續(xù)的業(yè)務(wù)邏輯

• 特點(diǎn)：由于木桶內(nèi)只要有令牌，請(qǐng)求就可以被處理，所以令牌桶算法可以支持突發(fā)流量。同時(shí)由于往木桶添加令牌的速度是固定的，且木桶的容量有上限，所以單位時(shí)間內(nèi)處理的請(qǐng)求書(shū)也能夠得到控制，起到限流的目的。假設(shè)加入令牌的速度為 1token/10ms，桶的容量為500，在請(qǐng)求比較的少的時(shí)候（小于每10毫秒1個(gè)請(qǐng)求）時(shí)，木桶可以先"攢"一些令牌（最多500個(gè)）。當(dāng)有突發(fā)流量時(shí)，一下把木桶內(nèi)的令牌取空，也就是有500個(gè)在并發(fā)執(zhí)行的業(yè)務(wù)邏輯，之后要等每10ms補(bǔ)充一個(gè)新的令牌才能接收一個(gè)新的請(qǐng)求。

• 參數(shù)設(shè)置：木桶的容量 - 考慮業(yè)務(wù)邏輯的資源消耗和機(jī)器能承載并發(fā)處理多少業(yè)務(wù)邏輯。生成令牌的速度 - 太慢的話起不到“攢”令牌應(yīng)對(duì)突發(fā)流量的效果。

• 適用場(chǎng)景：
  適合電商搶購(gòu)或者微博出現(xiàn)熱點(diǎn)事件這種場(chǎng)景，因?yàn)樵谙蘖鞯耐瑫r(shí)可以應(yīng)對(duì)一定的突發(fā)流量。如果采用均勻速度處理請(qǐng)求的算法，在發(fā)生熱點(diǎn)時(shí)間的時(shí)候，會(huì)造成大量的用戶無(wú)法訪問(wèn)，對(duì)用戶體驗(yàn)的損害比較大。

• go語(yǔ)言實(shí)現(xiàn)：
  假設(shè)每100ms生產(chǎn)一個(gè)令牌，按user_id/IP記錄訪問(wèn)最近一次訪問(wèn)的時(shí)間戳 t_last 和令牌數(shù)，每次請(qǐng)求時(shí)如果 now - last > 100ms, 增加 (now - last) / 100ms個(gè)令牌。然后，如果令牌數(shù) > 0，令牌數(shù) -1 繼續(xù)執(zhí)行后續(xù)的業(yè)務(wù)邏輯，否則返回請(qǐng)求頻率超限的錯(cuò)誤碼或頁(yè)面。

package main
import ( "fmt" "sync" "time")
// 并發(fā)訪問(wèn)同一個(gè)user_id/ip的記錄需要上鎖var recordMu map[string]*sync.RWMutex
func init() { recordMu = make(map[string]*sync.RWMutex)}
func max(a, b int) int { if a > b {  return a } return b}
type TokenBucket struct { BucketSize int // 木桶內(nèi)的容量：最多可以存放多少個(gè)令牌 TokenRate time.Duration // 多長(zhǎng)時(shí)間生成一個(gè)令牌 records map[string]*record // 報(bào)錯(cuò)user_id/ip的訪問(wèn)記錄}
// 上次訪問(wèn)時(shí)的時(shí)間戳和令牌數(shù)type record struct { last time.Time token int}
func NewTokenBucket(bucketSize int, tokenRate time.Duration) *TokenBucket { return &TokenBucket{  BucketSize: bucketSize,  TokenRate:  tokenRate,  records:    make(map[string]*record), }}
func (t *TokenBucket) getUidOrIp() string { // 獲取請(qǐng)求用戶的user_id或者ip地址 return "127.0.0.1"}
// 獲取這個(gè)user_id/ip上次訪問(wèn)時(shí)的時(shí)間戳和令牌數(shù)func (t *TokenBucket) getRecord(uidOrIp string) *record { if r, ok := t.records[uidOrIp]; ok {  return r } return &record{}}
// 保存user_id/ip最近一次請(qǐng)求時(shí)的時(shí)間戳和令牌數(shù)量func (t *TokenBucket) storeRecord(uidOrIp string, r *record) { t.records[uidOrIp] = r}
// 驗(yàn)證是否能獲取一個(gè)令牌func (t *TokenBucket) validate(uidOrIp string) bool { // 并發(fā)修改同一個(gè)用戶的記錄上寫(xiě)鎖 rl, ok := recordMu[uidOrIp] if !ok {  var mu sync.RWMutex  rl = &mu  recordMu[uidOrIp] = rl } rl.Lock() defer rl.Unlock()
 r := t.getRecord(uidOrIp) now := time.Now() if r.last.IsZero() {  // 第一次訪問(wèn)初始化為最大令牌數(shù)  r.last, r.token = now, t.BucketSize } else {  if r.last.Add(t.TokenRate).Before(now) {   // 如果與上次請(qǐng)求的間隔超過(guò)了token rate   // 則增加令牌，更新last   r.token += max(int(now.Sub(r.last) / t.TokenRate), t.BucketSize)   r.last = now  } } var result bool if r.token > 0 {  // 如果令牌數(shù)大于1，取走一個(gè)令牌，validate結(jié)果為true  r.token--  result = true }
 // 保存最新的record t.storeRecord(uidOrIp, r) return result}
// 返回是否被限流func (t *TokenBucket) IsLimited() bool { return !t.validate(t.getUidOrIp())}
func main() { tokenBucket := NewTokenBucket(5, 100*time.Millisecond) for i := 0; i< 6; i++ {  fmt.Println(tokenBucket.IsLimited()) } time.Sleep(100 * time.Millisecond) fmt.Println(tokenBucket.IsLimited())}

三、滑動(dòng)時(shí)間窗口算法

• 算法思想
  滑動(dòng)時(shí)間窗口算法，是從對(duì)普通時(shí)間窗口計(jì)數(shù)的優(yōu)化。
  使用普通時(shí)間窗口時(shí)，我們會(huì)為每個(gè)user_id/ip維護(hù)一個(gè)KV: uidOrIp: timestamp_requestCount。假設(shè)限制1秒1000個(gè)請(qǐng)求，那么第100ms有一個(gè)請(qǐng)求，這個(gè)KV變成 uidOrIp: timestamp_1，遞200ms有1個(gè)請(qǐng)求，我們先比較距離記錄的timestamp有沒(méi)有超過(guò)1s，如果沒(méi)有只更新count，此時(shí)KV變成 uidOrIp: timestamp_2。當(dāng)?shù)?100ms來(lái)一個(gè)請(qǐng)求時(shí)，更新記錄中的timestamp并重置計(jì)數(shù)，KV變成 uidOrIp: newtimestamp_1
  普通時(shí)間窗口有一個(gè)問(wèn)題，假設(shè)有500個(gè)請(qǐng)求集中在前1s的后100ms，500個(gè)請(qǐng)求集中在后1s的前100ms，其實(shí)在這200ms沒(méi)就已經(jīng)請(qǐng)求超限了，但是由于時(shí)間窗每經(jīng)過(guò)1s就會(huì)重置計(jì)數(shù)，就無(wú)法識(shí)別到此時(shí)的請(qǐng)求超限。
  對(duì)于滑動(dòng)時(shí)間窗口，我們可以把1ms的時(shí)間窗口劃分成10個(gè)time slot, 每個(gè)time slot統(tǒng)計(jì)某個(gè)100ms的請(qǐng)求數(shù)量。每經(jīng)過(guò)100ms，有一個(gè)新的time slot加入窗口，早于當(dāng)前時(shí)間100ms的time slot出窗口。窗口內(nèi)最多維護(hù)10個(gè)time slot，儲(chǔ)存空間的消耗同樣是比較低的。

• 適用場(chǎng)景
  與令牌桶一樣，有應(yīng)對(duì)突發(fā)流量的能力

• go語(yǔ)言實(shí)現(xiàn)
  主要就是實(shí)現(xiàn)sliding window算法?？梢詤⒖糂ilibili開(kāi)源的kratos框架里circuit breaker用循環(huán)列表保存time slot對(duì)象的實(shí)現(xiàn)，他們這個(gè)實(shí)現(xiàn)的好處是不用頻繁的創(chuàng)建和銷(xiāo)毀time slot對(duì)象。下面給出一個(gè)簡(jiǎn)單的基本實(shí)現(xiàn)：

package main
import ( "fmt" "sync" "time")
var winMu map[string]*sync.RWMutex
func init() { winMu = make(map[string]*sync.RWMutex)}
type timeSlot struct { timestamp time.Time // 這個(gè)timeSlot的時(shí)間起點(diǎn) count     int       // 落在這個(gè)timeSlot內(nèi)的請(qǐng)求數(shù)}
func countReq(win []*timeSlot) int { var count int for _, ts := range win {  count += ts.count } return count}
type SlidingWindowLimiter struct { SlotDuration time.Duration // time slot的長(zhǎng)度 WinDuration  time.Duration // sliding window的長(zhǎng)度 numSlots     int           // window內(nèi)最多有多少個(gè)slot windows      map[string][]*timeSlot maxReq       int // win duration內(nèi)允許的最大請(qǐng)求數(shù)}
func NewSliding(slotDuration time.Duration, winDuration time.Duration, maxReq int) *SlidingWindowLimiter { return &SlidingWindowLimiter{  SlotDuration: slotDuration,  WinDuration:  winDuration,  numSlots:     int(winDuration / slotDuration),  windows:      make(map[string][]*timeSlot),  maxReq:       maxReq, }}
// 獲取user_id/ip的時(shí)間窗口func (l *SlidingWindowLimiter) getWindow(uidOrIp string) []*timeSlot { win, ok := l.windows[uidOrIp] if !ok {  win = make([]*timeSlot, 0, l.numSlots) } return win}
func (l *SlidingWindowLimiter) storeWindow(uidOrIp string, win []*timeSlot) { l.windows[uidOrIp] = win}
func (l *SlidingWindowLimiter) validate(uidOrIp string) bool { // 同一user_id/ip并發(fā)安全 mu, ok := winMu[uidOrIp] if !ok {  var m sync.RWMutex  mu = &m  winMu[uidOrIp] = mu } mu.Lock() defer mu.Unlock()
 win := l.getWindow(uidOrIp) now := time.Now() // 已經(jīng)過(guò)期的time slot移出時(shí)間窗 timeoutOffset := -1 for i, ts := range win {  if ts.timestamp.Add(l.WinDuration).After(now) {   break  }  timeoutOffset = i } if timeoutOffset > -1 {  win = win[timeoutOffset+1:] }
 // 判斷請(qǐng)求是否超限 var result bool if countReq(win) < l.maxReq {  result = true }
 // 記錄這次的請(qǐng)求數(shù) var lastSlot *timeSlot if len(win) > 0 {  lastSlot = win[len(win)-1]  if lastSlot.timestamp.Add(l.SlotDuration).Before(now) {   lastSlot = &timeSlot{timestamp: now, count: 1}   win = append(win, lastSlot)  } else {   lastSlot.count++  } } else {  lastSlot = &timeSlot{timestamp: now, count: 1}  win = append(win, lastSlot) }
 l.storeWindow(uidOrIp, win)
 return result}
func (l *SlidingWindowLimiter) getUidOrIp() string { return "127.0.0.1"}
func (l *SlidingWindowLimiter) IsLimited() bool { return !l.validate(l.getUidOrIp())}
func main() { limiter := NewSliding(100*time.Millisecond, time.Second, 10) for i := 0; i < 5; i++ {  fmt.Println(limiter.IsLimited()) } time.Sleep(100 * time.Millisecond) for i := 0; i < 5; i++ {  fmt.Println(limiter.IsLimited()) } fmt.Println(limiter.IsLimited()) for _, v := range limiter.windows[limiter.getUidOrIp()] {  fmt.Println(v.timestamp, v.count) }
 fmt.Println("a thousand years later...") time.Sleep(time.Second) for i := 0; i < 7; i++ {  fmt.Println(limiter.IsLimited()) } for _, v := range limiter.windows[limiter.getUidOrIp()] {  fmt.Println(v.timestamp, v.count) }}