幾種限流算法的go語言實(shí)現(xiàn)

不依賴外部庫的情況下,限流算法有什么實(shí)現(xiàn)的思路?本文介紹了3種實(shí)現(xiàn)限流的方式。
一、漏桶算法
算法思想 與令牌桶是“反向”的算法,當(dāng)有請求到來時先放到木桶中,worker以固定的速度從木桶中取出請求進(jìn)行相應(yīng)。如果木桶已經(jīng)滿了,直接返回請求頻率超限的錯誤碼或者頁面 適用場景
流量最均勻的限流方式,一般用于流量“整形”,例如保護(hù)數(shù)據(jù)庫的限流。先把對數(shù)據(jù)庫的訪問加入到木桶中,worker再以db能夠承受的qps從木桶中取出請求,去訪問數(shù)據(jù)庫。不太適合電商搶購和微博出現(xiàn)熱點(diǎn)事件等場景的限流,一是應(yīng)對突發(fā)流量不是很靈活,二是為每個user_id/ip維護(hù)一個隊列(木桶),workder從這些隊列中拉取任務(wù),資源的消耗會比較大。
go語言實(shí)現(xiàn)
通常使用隊列來實(shí)現(xiàn),在go語言中可以通過buffered channel來快速實(shí)現(xiàn),任務(wù)加入channel,開啟一定數(shù)量的worker從channel中獲取任務(wù)執(zhí)行。
package mainimport ("fmt""sync""time")// 每個請求來了,把需要執(zhí)行的業(yè)務(wù)邏輯封裝成Task,放入木桶,等待worker取出執(zhí)行type Task struct {handler func() Result // worker從木桶中取出請求對象后要執(zhí)行的業(yè)務(wù)邏輯函數(shù)resChan chan Result // 等待worker執(zhí)行并返回結(jié)果的channeltaskID 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 // 同時從木桶中獲取任務(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)滿了,返回falseselect {case b.bucket <- task:default:fmt.Printf("request[id=%d] is refused\n", task.taskID)return false}// 等待worker執(zhí)行<-task.resChanfmt.Printf("request[id=%d] is run\n", task.taskID)return true}func (b *LeakyBucket) Start() {// 開啟worker從木桶拉取任務(wù)執(zhí)行go func() {for i := 0; i < b.NumWorker; i++ {go func() {for {task := <-b.bucketresult := task.handler()task.resChan <- result}}()}}()}func main() {bucket := NewLeakyBucket(10, 4)bucket.Start()var wg sync.WaitGroupfor i := 0; i < 20; i++ {wg.Add(1)go func(id int) {defer wg.Done()task := NewTask(id)bucket.validate(task)}(i)}wg.Wait()}
二、令牌桶算法
算法思想
想象有一個木桶,以固定的速度往木桶里加入令牌,木桶滿了則不再加入令牌。服務(wù)收到請求時嘗試從木桶中取出一個令牌,如果能夠得到令牌則繼續(xù)執(zhí)行后續(xù)的業(yè)務(wù)邏輯;如果沒有得到令牌,直接返回反問頻率超限的錯誤碼或頁面等,不繼續(xù)執(zhí)行后續(xù)的業(yè)務(wù)邏輯
特點(diǎn):由于木桶內(nèi)只要有令牌,請求就可以被處理,所以令牌桶算法可以支持突發(fā)流量。同時由于往木桶添加令牌的速度是固定的,且木桶的容量有上限,所以單位時間內(nèi)處理的請求書也能夠得到控制,起到限流的目的。假設(shè)加入令牌的速度為 1token/10ms,桶的容量為500,在請求比較的少的時候(小于每10毫秒1個請求)時,木桶可以先"攢"一些令牌(最多500個)。當(dāng)有突發(fā)流量時,一下把木桶內(nèi)的令牌取空,也就是有500個在并發(fā)執(zhí)行的業(yè)務(wù)邏輯,之后要等每10ms補(bǔ)充一個新的令牌才能接收一個新的請求。
參數(shù)設(shè)置:木桶的容量 - 考慮業(yè)務(wù)邏輯的資源消耗和機(jī)器能承載并發(fā)處理多少業(yè)務(wù)邏輯。生成令牌的速度 - 太慢的話起不到“攢”令牌應(yīng)對突發(fā)流量的效果。
適用場景:
適合電商搶購或者微博出現(xiàn)熱點(diǎn)事件這種場景,因為在限流的同時可以應(yīng)對一定的突發(fā)流量。如果采用均勻速度處理請求的算法,在發(fā)生熱點(diǎn)時間的時候,會造成大量的用戶無法訪問,對用戶體驗的損害比較大。
go語言實(shí)現(xiàn):
假設(shè)每100ms生產(chǎn)一個令牌,按user_id/IP記錄訪問最近一次訪問的時間戳 t_last 和令牌數(shù),每次請求時如果 now - last > 100ms, 增加 (now - last) / 100ms個令牌。然后,如果令牌數(shù) > 0,令牌數(shù) -1 繼續(xù)執(zhí)行后續(xù)的業(yè)務(wù)邏輯,否則返回請求頻率超限的錯誤碼或頁面。
package mainimport ("fmt""sync""time")// 并發(fā)訪問同一個user_id/ip的記錄需要上鎖var recordMu map[string]*sync.RWMutexfunc 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)的容量:最多可以存放多少個令牌TokenRate time.Duration // 多長時間生成一個令牌records map[string]*record // 報錯user_id/ip的訪問記錄}// 上次訪問時的時間戳和令牌數(shù)type record struct {last time.Timetoken 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 {// 獲取請求用戶的user_id或者ip地址return "127.0.0.1"}// 獲取這個user_id/ip上次訪問時的時間戳和令牌數(shù)func (t *TokenBucket) getRecord(uidOrIp string) *record {if r, ok := t.records[uidOrIp]; ok {return r}return &record{}}// 保存user_id/ip最近一次請求時的時間戳和令牌數(shù)量func (t *TokenBucket) storeRecord(uidOrIp string, r *record) {t.records[uidOrIp] = r}// 驗證是否能獲取一個令牌func (t *TokenBucket) validate(uidOrIp string) bool {// 并發(fā)修改同一個用戶的記錄上寫鎖rl, ok := recordMu[uidOrIp]if !ok {var mu sync.RWMutexrl = &murecordMu[uidOrIp] = rl}rl.Lock()defer rl.Unlock()r := t.getRecord(uidOrIp)now := time.Now()if r.last.IsZero() {// 第一次訪問初始化為最大令牌數(shù)r.last, r.token = now, t.BucketSize} else {if r.last.Add(t.TokenRate).Before(now) {// 如果與上次請求的間隔超過了token rate// 則增加令牌,更新lastr.token += max(int(now.Sub(r.last) / t.TokenRate), t.BucketSize)r.last = now}}var result boolif r.token > 0 {// 如果令牌數(shù)大于1,取走一個令牌,validate結(jié)果為truer.token--result = true}// 保存最新的recordt.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())}
三、滑動時間窗口算法
算法思想
滑動時間窗口算法,是從對普通時間窗口計數(shù)的優(yōu)化。
使用普通時間窗口時,我們會為每個user_id/ip維護(hù)一個KV: uidOrIp: timestamp_requestCount。假設(shè)限制1秒1000個請求,那么第100ms有一個請求,這個KV變成 uidOrIp: timestamp_1,遞200ms有1個請求,我們先比較距離記錄的timestamp有沒有超過1s,如果沒有只更新count,此時KV變成 uidOrIp: timestamp_2。當(dāng)?shù)?100ms來一個請求時,更新記錄中的timestamp并重置計數(shù),KV變成 uidOrIp: newtimestamp_1
普通時間窗口有一個問題,假設(shè)有500個請求集中在前1s的后100ms,500個請求集中在后1s的前100ms,其實(shí)在這200ms沒就已經(jīng)請求超限了,但是由于時間窗每經(jīng)過1s就會重置計數(shù),就無法識別到此時的請求超限。
對于滑動時間窗口,我們可以把1ms的時間窗口劃分成10個time slot, 每個time slot統(tǒng)計某個100ms的請求數(shù)量。每經(jīng)過100ms,有一個新的time slot加入窗口,早于當(dāng)前時間100ms的time slot出窗口。窗口內(nèi)最多維護(hù)10個time slot,儲存空間的消耗同樣是比較低的。
適用場景
與令牌桶一樣,有應(yīng)對突發(fā)流量的能力
go語言實(shí)現(xiàn)
主要就是實(shí)現(xiàn)sliding window算法。可以參考Bilibili開源的kratos框架里circuit breaker用循環(huán)列表保存time slot對象的實(shí)現(xiàn),他們這個實(shí)現(xiàn)的好處是不用頻繁的創(chuàng)建和銷毀time slot對象。下面給出一個簡單的基本實(shí)現(xiàn):
package mainimport ("fmt""sync""time")var winMu map[string]*sync.RWMutexfunc init() {winMu = make(map[string]*sync.RWMutex)}type timeSlot struct {timestamp time.Time // 這個timeSlot的時間起點(diǎn)count int // 落在這個timeSlot內(nèi)的請求數(shù)}func countReq(win []*timeSlot) int {var count intfor _, ts := range win {count += ts.count}return count}type SlidingWindowLimiter struct {SlotDuration time.Duration // time slot的長度WinDuration time.Duration // sliding window的長度numSlots int // window內(nèi)最多有多少個slotwindows map[string][]*timeSlotmaxReq int // win duration內(nèi)允許的最大請求數(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的時間窗口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.RWMutexmu = &mwinMu[uidOrIp] = mu}mu.Lock()defer mu.Unlock()win := l.getWindow(uidOrIp)now := time.Now()// 已經(jīng)過期的time slot移出時間窗timeoutOffset := -1for i, ts := range win {if ts.timestamp.Add(l.WinDuration).After(now) {break}timeoutOffset = i}if timeoutOffset > -1 {win = win[timeoutOffset+1:]}// 判斷請求是否超限var result boolif countReq(win) < l.maxReq {result = true}// 記錄這次的請求數(shù)var lastSlot *timeSlotif 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)}}
鏈接:https://juejin.cn/post/6844904051344146439
(版權(quán)歸原作者所有,侵刪)
評論
圖片
表情
