大家好，我是dandyhuang。高并發(fā)三板斧：限流、緩存、降級(jí)。 限流其實(shí)就是：當(dāng)高并發(fā)或者瞬時(shí)高并發(fā)時(shí)，為了保證系統(tǒng)的穩(wěn)定性、可用性，系統(tǒng)以犧牲部分請(qǐng)求為代價(jià)或者延遲處理請(qǐng)求為代價(jià)，保證系統(tǒng)整體服務(wù)可用。今天就給大家介紹golang 官方擴(kuò)展包 time(golang.org/x/time/rate) 中，一個(gè)基于令牌桶的限流器實(shí)現(xiàn)。它的實(shí)現(xiàn)java guava rate limiter中的實(shí)現(xiàn)思路是一樣的。按照該思路我也用c++實(shí)現(xiàn)了一份代碼，可以供大家參考代碼token_limiter.cpp實(shí)現(xiàn)。

令牌桶

令牌桶.png

令牌桶算法的原理：系統(tǒng)會(huì)以一個(gè)恒定的速度往桶里放入令牌，而如果請(qǐng)求需要被處理，則需要先從桶里獲取一個(gè)令牌，當(dāng)桶里沒(méi)有令牌可取時(shí)，則拒絕服務(wù)。自然我們可能會(huì)想到啟一個(gè)協(xié)程，定時(shí)的往桶里丟入令牌。但golang中沒(méi)有按照這個(gè)思路去實(shí)現(xiàn)。而是實(shí)時(shí)計(jì)算當(dāng)前產(chǎn)生的令牌數(shù)?？赡軙?huì)有人覺(jué)得，把定時(shí)的時(shí)間片調(diào)低一些，比如1us就去計(jì)算產(chǎn)生的令牌數(shù)。這種效果也和實(shí)時(shí)計(jì)算效果也差不多。但這種處理，隨著精度越高，cpu親緣性的問(wèn)題就越嚴(yán)重。我們來(lái)看看golang是怎么實(shí)現(xiàn)的。思路其實(shí)也是比較巧妙。

time/rate實(shí)現(xiàn)

package limiter

import (
 "fmt"
 "testing"
 "time"

 "golang.org/x/time/rate"
)

func TestLimter(t *testing.T) {
    limiter := rate.NewLimiter(rate.Every(time.Millisecond*10), 2)
    for i := 0; i < 10; i++ {
        var ok bool
        if limiter.Allow() {
            ok = true
        }
        time.Sleep(time.Millisecond * 3)
        fmt.Println(ok, limiter.Burst())
    }
}

執(zhí)行結(jié)果：

=== RUN   TestLimter
true 2
true 2
false 2
true 2
false 2
false 2
true 2
false 2
false 2
true 2
--- PASS: TestLimter1 (0.03s)

因?yàn)槌跏蓟?個(gè)令牌在里頭，隨著后續(xù)執(zhí)行，每10ms產(chǎn)生一個(gè)令牌。所以后續(xù)10ms內(nèi)，只有一個(gè)請(qǐng)求能獲取到令牌。

創(chuàng)建限流器

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

type Limiter struct {
    mu     sync.Mutex
    limit  Limit
    burst  int
    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
}

Limiter參數(shù)中

• mu：每次請(qǐng)求并發(fā)安全，加鎖處理

• burst : 桶內(nèi)能存放令牌的個(gè)數(shù)

• limit：生成令牌的速率

• tokens： 剩余令牌個(gè)數(shù)

• last： 上一次取走 token 的時(shí)間

• lastEvent：上一次限流事件的時(shí)間

Allow 判斷是否運(yùn)行通過(guò)

// 判斷是否滿足條件
func (lim *Limiter) Allow() bool {
    return lim.AllowN(time.Now(), 1)
}

func (lim *Limiter) AllowN(now time.Time, n int) bool {
    return lim.reserveN(now, n, 0).ok
}

func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation {
  // 加鎖防止并發(fā)
    lim.mu.Lock()
    defer lim.mu.Unlock()
    // 邊界處理，都是一些異常設(shè)置，設(shè)置最大值，或沒(méi)有設(shè)置的場(chǎng)景
    if lim.limit == Inf {
        return Reservation{
            ok:        true,
            lim:       lim,
            tokens:    n,
            timeToAct: now,
        }
    } else if lim.limit == 0 {
        var ok bool
        if lim.burst >= n {
            ok = true
            lim.burst -= n
        }
        return Reservation{
            ok:        ok,
            lim:       lim,
            tokens:    lim.burst,
            timeToAct: now,
        }
    }
     // 從上次獲取令牌到當(dāng)前時(shí)間，共產(chǎn)生的令牌數(shù)tokens
    now, last, tokens := lim.advance(now)

    // 減去需要的令牌數(shù)據(jù)
    tokens -= float64(n)

    var waitDuration time.Duration
  // 如果需要的令牌數(shù)tokens 為負(fù)數(shù)，則計(jì)算需要等待的 WaitDuration時(shí)間
    if tokens < 0 {
        waitDuration = lim.limit.durationFromTokens(-tokens)
    }

    // 獲取需要的令牌數(shù)如果大于桶的容量  或者 等待時(shí)間大于設(shè)置的maxFutureReserve
    ok := n <= lim.burst && waitDuration <= maxFutureReserve

    // 構(gòu)造一個(gè)Reservation對(duì)象，后續(xù)結(jié)果返回該對(duì)象
    r := Reservation{
        ok:    ok,
        lim:   lim,
        limit: lim.limit,
    }
  // 成功才更新該對(duì)象
    if ok {
        r.tokens = n
        r.timeToAct = now.Add(waitDuration)
    }

    // 成功了，才把上次時(shí)間和tokens數(shù)更新
    if ok {
        lim.last = now
        lim.tokens = tokens
        lim.lastEvent = r.timeToAct
    } else {
    // 這里因?yàn)殒i并發(fā)的問(wèn)題，所以才回導(dǎo)致last會(huì)有更新
        lim.last = last
    }

    return r
}

reserveN中記錄了上次訪問(wèn)的時(shí)間和當(dāng)前桶中令牌的數(shù)量。當(dāng)再次訪問(wèn)時(shí)，通過(guò)上次訪問(wèn)記錄的時(shí)間實(shí)時(shí)計(jì)算當(dāng)前令牌的數(shù)量，決定是否可以放行。

advance計(jì)算產(chǎn)生的令牌數(shù)

func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
    last := lim.last
  // 因?yàn)榧渔i的原因，所以出現(xiàn)這種情況
    if now.Before(last) {
        last = now
    }

    // （當(dāng)前時(shí)間-上次時(shí)間）* 速率 = 產(chǎn)生的令牌數(shù)
    elapsed := now.Sub(last)
    delta := lim.limit.tokensFromDuration(elapsed)
    tokens := lim.tokens + delta
    if burst := float64(lim.burst); tokens > burst {
        tokens = burst
    }
    return now, last, tokens
}

這里AllowN()請(qǐng)求中，我們是在Allow()的時(shí)候調(diào)用lim.AllowN(time.Now(), 1)，并把當(dāng)前時(shí)間傳入。這時(shí)候，如果

reserveN處理的比較慢，并且請(qǐng)求成功。如線程t1請(qǐng)求的時(shí)間為10:10:12秒，線程t2請(qǐng)求時(shí)間為10:10:10秒。 而線程t1先搶到了鎖。并處理請(qǐng)求成功。接下去t2才進(jìn)行處理。這時(shí)last為10:10:12秒，now為10:10:10秒的場(chǎng)景

總結(jié)

golang/rate包中，犧牲一點(diǎn)加鎖的性能，實(shí)時(shí)計(jì)算產(chǎn)生的令牌數(shù)。這種實(shí)現(xiàn)的好處： 對(duì)令牌的計(jì)算可以非常精確。而對(duì)比于定時(shí)往桶里添加令牌的實(shí)現(xiàn)，雖然在請(qǐng)求可以使用原子計(jì)算，不上鎖實(shí)現(xiàn)。但對(duì)于令牌的計(jì)算來(lái)說(shuō)，是比較不準(zhǔn)確的，需要根據(jù)定時(shí)器的精度來(lái)保證。而精度越小，cpu親緣性問(wèn)題就越明顯。個(gè)人覺(jué)得雖然加鎖的實(shí)現(xiàn)，對(duì)性能有一部分影響，但是令牌桶都是在計(jì)算，所以性能不會(huì)有很大的問(wèn)題，加鎖時(shí)間不長(zhǎng)。