限流器模型

golang.org/x/time/rate 限流器目前提供了一種令牌桶算法的的限流器。

請求需要拿到令牌才能接著往下執(zhí)行, 邏輯上有一個(gè)令牌桶，桶的最大容量是固定的。

當(dāng)桶內(nèi)令牌數(shù) 小于 桶的最大容量時(shí), 以固定的頻率向桶內(nèi)增加令牌直至令牌數(shù)滿。

每個(gè)請求理論上消耗一個(gè)令牌(實(shí)際上提供了的方法每次可以消耗大于一個(gè)的令牌)

限流器初始化時(shí) 令牌桶是滿的

簡單例子

package main

import (
    "fmt"
    "golang.org/x/time/rate"
    "time"
)

func main() {
    limiter := rate.NewLimiter(rate.Every(time.Millisecond * 31), 2)
    //time.Sleep(time.Second)
    for i := 0; i < 10; i++ {
        var ok bool
        if limiter.Allow() {
            ok = true
        }
        time.Sleep(time.Millisecond * 20)
        fmt.Println(ok, limiter.Burst())
    }
}

輸出

true 2
true 2
true 2
true 2
false 2
true 2
true 2
false 2
true 2
true 2

可以看出 一開始桶內(nèi)令牌數(shù)是滿的。 由于生成令牌的間隔比請求的間隔多了11ms, 所以到后面每兩個(gè)請求后就會(huì)失敗一次。

限流器實(shí)現(xiàn)原理

限流器的數(shù)據(jù)結(jié)構(gòu)和New函數(shù)如下:

type Limiter struct {
    limit Limit
    burst int

    mu     sync.Mutex
    tokens float64
    // last is the last time the limiter's tokens field was updated
    last time.Time
    // lastEvent is the latest time of a rate-limited event (past or future)
    lastEvent time.Time
}

func NewLimiter(r Limit, b int) *Limiter {
    return &Limiter{
        limit: r,
        burst: b,
    }
}

可見, 雖然在邏輯上, 令牌桶在沒滿的情況下, 是不斷往里面以一定時(shí)間間隔添加令牌, 但代碼實(shí)現(xiàn)上并沒有這樣的一個(gè)協(xié)程。 實(shí)際上桶剩余令牌的更新是推遲在每次消費(fèi)的時(shí)候進(jìn)行計(jì)算的。

核心函數(shù)

func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation

當(dāng)調(diào)用 limiter.Allow()的時(shí)候, 底層調(diào)用實(shí)際即使該函數(shù), 入?yún)ow=當(dāng)前時(shí)間, n=1, maxFutrueReserve=0

maxFutrueReserve在Wait()方法內(nèi)調(diào)用時(shí), 值會(huì)大于0。

函數(shù)前半段: 當(dāng)limit變量無限大(該變量是新增令牌的時(shí)間間隔的倒數(shù), 無限大即表示無間隔), 直接返回TRUE即可

    lim.mu.Lock()

    if lim.limit == Inf {
        lim.mu.Unlock()
        return Reservation{
            ok:        true,
            lim:       lim,
            tokens:    n,
            timeToAct: now,
        }
    }

之調(diào)用lim.advance(now) 函數(shù), 獲取上次令牌桶更新時(shí)間和此時(shí)桶內(nèi)令牌數(shù)應(yīng)該是多少(這一步就是上文提到的桶剩余令牌的延遲計(jì)算)

先檢查傳入的時(shí)間是否先于數(shù)據(jù)結(jié)構(gòu)內(nèi)記錄的上次令牌數(shù)更新時(shí)間(并發(fā)的情況下, now變量的初始化未在鎖內(nèi)執(zhí)行, 這種情況很有可能發(fā)生), 如果是則將上次令牌時(shí)更新時(shí)間替換為當(dāng)前時(shí)間
獲取上次令牌數(shù)更新時(shí)間到現(xiàn)在的時(shí)間間隔, 如果超過讓令牌桶加到滿的時(shí)間間隔，就使用讓令牌桶加到滿的時(shí)間間隔。這個(gè)變量用于計(jì)算增加令牌桶的令牌數(shù), 由于有容量限制，過大沒有意義
計(jì)算令牌桶此時(shí)的個(gè)數(shù), 返回

// advance calculates and returns an updated state for lim resulting from the passage of time.
// lim is not changed.
func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
    last := lim.last
    if now.Before(last) {
        last = now
    }

    // Avoid making delta overflow below when last is very old.
    maxElapsed := lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)
    elapsed := now.Sub(last)
    if elapsed > maxElapsed {
        elapsed = maxElapsed
    }

    // Calculate the new number of tokens, due to time that passed.
    delta := lim.limit.tokensFromDuration(elapsed)
    tokens := lim.tokens + delta
    if burst := float64(lim.burst); tokens > burst {
        tokens = burst
    }

    return now, last, tokens
}

函數(shù)后半段:

計(jì)算這次消耗的令牌數(shù), 但是不是直接簡單比較消耗后剩余令牌數(shù)大于0, 而是計(jì)算還需要多少時(shí)間令牌桶能生成滿足需要的令牌數(shù)(如果消耗后令牌數(shù)仍大于0 則需要時(shí)間就是0)。 將該值和入?yún)axFutureReserve比較再得出結(jié)果。即當(dāng)前令牌數(shù)可能是負(fù)的。 這么做的原因是限流器還提供了Wait方法, 當(dāng)前令牌不足時(shí)可以阻塞等待至有令牌為止(提前消費(fèi)令牌 然后阻塞一段時(shí)間)， 而不是像Allow方法一樣立即返回False

   // Calculate the remaining number of tokens resulting from the request.
    tokens -= float64(n)

    // Calculate the wait duration
    var waitDuration time.Duration
    if tokens < 0 {
        waitDuration = lim.limit.durationFromTokens(-tokens)
    }

    // Decide result
    ok := n <= lim.burst && waitDuration <= maxFutureReserve

    // Prepare reservation
    r := Reservation{
        ok:    ok,
        lim:   lim,
        limit: lim.limit,
    }
    if ok {
        r.tokens = n
      	r.timeToAct = now.Add(waitDuration)
    }

    // Update state
    if ok {
        lim.last = now
        lim.tokens = tokens
        lim.lastEvent = r.timeToAct
    } else {
        lim.last = last
    }

    lim.mu.Unlock()
    return r
}

總結(jié)

以上是生活随笔為你收集整理的golang官方包限流器使用和原理(golang.org/x/time/rate)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。