限流器是后臺(tái)服務(wù)中十分重要的組件，在實(shí)際的業(yè)務(wù)場(chǎng)景中使用居多，其設(shè)計(jì)在微服務(wù)、網(wǎng)關(guān)、和一些后臺(tái)服務(wù)中會(huì)經(jīng)常遇到。限流器的作用是用來限制其請(qǐng)求的速率，保護(hù)后臺(tái)響應(yīng)服務(wù)，以免服務(wù)過載導(dǎo)致服務(wù)不可用現(xiàn)象出現(xiàn)。

限流器的實(shí)現(xiàn)方法有很多種，例如 Token Bucket、滑動(dòng)窗口法、Leaky Bucket等。

在 Golang 庫(kù)中官方給我們提供了限流器的實(shí)現(xiàn)golang.org/x/time/rate，它是基于令牌桶算法（Token Bucket）設(shè)計(jì)實(shí)現(xiàn)的。

令牌桶算法

令牌桶設(shè)計(jì)比較簡(jiǎn)單，可以簡(jiǎn)單的理解成一個(gè)只能存放固定數(shù)量雪糕?的一個(gè)冰箱，每個(gè)請(qǐng)求可以理解成來拿雪糕的人，有且只能每一次請(qǐng)求拿一塊?，那雪糕拿完了會(huì)怎么樣呢？這里會(huì)有一個(gè)固定放雪糕的工人，并且他往冰箱里放雪糕的頻率都是一致的，例如他 1s 中只能往冰箱里放 10 塊雪糕，這里就可以看出請(qǐng)求響應(yīng)的頻率了。

令牌桶設(shè)計(jì)概念：

• 令牌：每次請(qǐng)求只有拿到 Token 令牌后，才可以繼續(xù)訪問；
• 桶：具有固定數(shù)量的桶，每個(gè)桶中最多只能放設(shè)計(jì)好的固定數(shù)量的令牌；
• 入桶頻率：按照固定的頻率往桶中放入令牌，放入令牌不能超過桶的容量。

也就是說，基于令牌桶設(shè)計(jì)算法就限制了請(qǐng)求的速率，達(dá)到請(qǐng)求響應(yīng)可控的目的，特別是針對(duì)于高并發(fā)場(chǎng)景中突發(fā)流量請(qǐng)求的現(xiàn)象，后臺(tái)就可以輕松應(yīng)對(duì)請(qǐng)求了，因?yàn)榈胶蠖司唧w服務(wù)的時(shí)候突發(fā)流量請(qǐng)求已經(jīng)經(jīng)過了限流了。

具體設(shè)計(jì)

限流器定義

type Limiter struct {
	mu        sync.Mutex // 互斥鎖（排他鎖）
	limit     Limit      // 放入桶的頻率  float64 類型
	burst     int        // 桶的大小
	tokens    float64    // 令牌 token 當(dāng)前剩余的數(shù)量
	last      time.Time  // 最近取走 token 的時(shí)間
	lastEvent time.Time  // 最近限流事件的時(shí)間
}

limit、burst 和 token 是這個(gè)限流器中核心的參數(shù)，請(qǐng)求并發(fā)的大小在這里實(shí)現(xiàn)的。

在令牌發(fā)放之后，會(huì)存儲(chǔ)在 Reservation 預(yù)約對(duì)象中：

type Reservation struct {
	ok        bool      // 是否滿足條件分配了 token
	lim       *Limiter  // 發(fā)送令牌的限流器
	tokens    int       // 發(fā)送 token 令牌的數(shù)量
	timeToAct time.Time // 滿足令牌發(fā)放的時(shí)間
	limit     Limit     // 令牌發(fā)放速度
}

消費(fèi) Token

Limiter 提供了三類方法供用戶消費(fèi) Token，用戶可以每次消費(fèi)一個(gè) Token，也可以一次性消費(fèi)多個(gè) Token。而每種方法代表了當(dāng) Token 不足時(shí)，各自不同的對(duì)應(yīng)手段。

Wait、WaitN

func (lim *Limiter) Wait(ctx context.Context) (err error)
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)

其中，Wait 就是 WaitN(ctx, 1)，在下面的方法介紹實(shí)現(xiàn)也是一樣的。

使用 Wait 方法消費(fèi) Token 時(shí)，如果此時(shí)桶內(nèi) Token 數(shù)組不足 ( 小于 n )，那么 Wait 方法將會(huì)阻塞一段時(shí)間，直至 Token 滿足條件。如果充足則直接返回。

Allow、AllowN

func (lim *Limiter) Allow() bool
func (lim *Limiter) AllowN(now time.Time, n int) bool 

AllowN 方法表示，截止到當(dāng)前某一時(shí)刻，目前桶中數(shù)目是否至少為 n 個(gè)，滿足則返回 true，同時(shí)從桶中消費(fèi) n 個(gè) token。 反之返回不消費(fèi) Token，false。

通常對(duì)應(yīng)這樣的線上場(chǎng)景，如果請(qǐng)求速率過快，就直接丟到某些請(qǐng)求。

Reserve、ReserveN

官方提供的限流器有阻塞等待式的 Wait，也有直接判斷方式的 Allow，還有提供了自己維護(hù)預(yù)留式的，但核心的實(shí)現(xiàn)都是下面的 reserveN 方法。

func (lim *Limiter) Reserve() *Reservation
func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation

當(dāng)調(diào)用完成后，無論 Token 是否充足，都會(huì)返回一個(gè)Reservation *對(duì)象。

你可以調(diào)用該對(duì)象的 Delay() 方法，該方法返回了需要等待的時(shí)間。如果等待時(shí)間為 0，則說明不用等待。 必須等到等待時(shí)間結(jié)束之后，才能進(jìn)行接下來的工作。

或者，如果不想等待，可以調(diào)用 Cancel() 方法，該方法會(huì)將 Token 歸還。

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

	// 首先判斷是否放入頻率是否為無窮大
	// 如果為無窮大，說明暫時(shí)不限流
	if lim.limit == Inf {
		lim.mu.Unlock()
		return Reservation{
			ok:        true,
			lim:       lim,
			tokens:    n,
			timeToAct: now,
		}
	}

	// 拿到截至 now 時(shí)間時(shí)
	// 可以獲取的令牌 tokens 數(shù)量及上一次拿走令牌的時(shí)間 last
	now, last, tokens := lim.advance(now)

	// 更新 tokens 數(shù)量
	tokens -= float64(n)

	// 如果 tokens 為負(fù)數(shù)，代表當(dāng)前沒有 token 放入桶中
	// 說明需要等待，計(jì)算等待的時(shí)間
	var waitDuration time.Duration
	if tokens < 0 {
		waitDuration = lim.limit.durationFromTokens(-tokens)
	}

	// 計(jì)算是否滿足分配條件
	// 1、需要分配的大小不超過桶的大小
	// 2、等待時(shí)間不超過設(shè)定的等待時(shí)長(zhǎng)
	ok := n <= lim.burst && waitDuration <= maxFutureReserve

	// 預(yù)處理 reservation
	r := Reservation{
		ok:    ok,
		lim:   lim,
		limit: lim.limit,
	}
	// 若當(dāng)前滿足分配條件
	// 1、設(shè)置分配大小
	// 2、滿足令牌發(fā)放的時(shí)間 = 當(dāng)前時(shí)間 + 等待時(shí)長(zhǎng)
	if ok {
		r.tokens = n
		r.timeToAct = now.Add(waitDuration)
	}

	// 更新 limiter 的值，并返回
	if ok {
		lim.last = now
		lim.tokens = tokens
		lim.lastEvent = r.timeToAct
	} else {
		lim.last = last
	}

	lim.mu.Unlock()
	return r
}

具體使用

rate 包中提供了對(duì)限流器的使用，只需要指定 limit（放入桶中的頻率）、burst（桶的大?。?。

func NewLimiter(r Limit, b int) *Limiter {
	return &Limiter{
		limit: r, // 放入桶的頻率
		burst: b, // 桶的大小
	}
}

在這里，使用一個(gè) http api 來簡(jiǎn)單的驗(yàn)證一下 time/rate 的強(qiáng)大：

func main() {
	r := rate.Every(1 * time.Millisecond)
	limit := rate.NewLimiter(r, 10)
	http.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
		if limit.Allow() {
			fmt.Printf("請(qǐng)求成功，當(dāng)前時(shí)間：%s\n", time.Now().Format("2006-01-02 15:04:05"))
		} else {
			fmt.Printf("請(qǐng)求成功，但是被限流了。。。\n")
		}
	})

	_ = http.ListenAndServe(":8081", nil)
}

在這里，我把桶設(shè)置成了每一毫秒投放一次令牌，桶容量大小為 10，起一個(gè) http 的服務(wù)，模擬后臺(tái) API。

接下來做一個(gè)壓力測(cè)試，看看效果如何：

func GetApi() {
	api := "http://localhost:8081/"
	res, err := http.Get(api)
	if err != nil {
		panic(err)
	}
	defer res.Body.Close()

	if res.StatusCode == http.StatusOK {
		fmt.Printf("get api success\n")
	}
}

func Benchmark_Main(b *testing.B) {
	for i := 0; i < b.N; i++ {
		GetApi()
	}
}

效果如下：

......
請(qǐng)求成功，當(dāng)前時(shí)間：2020-08-24 14:26:52
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，當(dāng)前時(shí)間：2020-08-24 14:26:52
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
請(qǐng)求成功，但是被限流了。。。
......

在這里，可以看到，當(dāng)使用 AllowN 方法中，只有當(dāng)令牌 Token 生產(chǎn)出來，才可以消費(fèi)令牌，繼續(xù)請(qǐng)求，剩余的則是將其請(qǐng)求拋棄，當(dāng)然在實(shí)際的業(yè)務(wù)處理中，可以用比較友好的方式反饋給前端。

在這里，先有的幾次請(qǐng)求都會(huì)成功，是因?yàn)榉?wù)啟動(dòng)后，令牌桶會(huì)初始化，將令牌放入到桶中，但是隨著突發(fā)流量的請(qǐng)求，令牌按照預(yù)定的速率生產(chǎn)令牌，就會(huì)出現(xiàn)明顯的令牌供不應(yīng)求的現(xiàn)象。

到此這篇關(guān)于Golang限流器time/rate設(shè)計(jì)與實(shí)現(xiàn)詳解的文章就介紹到這了,更多相關(guān)Go限流器time/rate內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論