前言

今天與大家聊一聊高并發(fā)系統(tǒng)中的限流技術(shù)，限流又稱為流量控制，是指限制到達系統(tǒng)的并發(fā)請求數(shù)，當(dāng)達到限制條件則可以拒絕請求，可以起到保護下游服務(wù)，防止服務(wù)過載等作用。常用的限流策略有漏桶算法、令牌桶算法、滑動窗口；下文主要與大家一起分析一下漏桶算法和令牌桶算法，滑動窗口就不在這里這介紹了。好啦，廢話不多話，開整。

文中測試代碼已上傳：github.com/asong2020/G… 

漏桶算法

漏桶算法比較好理解，假設(shè)我們現(xiàn)在有一個水桶，我們向這個水桶里添水，雖然我們我們無法預(yù)計一次會添多少水，也無法預(yù)計水流入的速度，但是可以固定出水的速度，不論添水的速率有多大，都按照固定的速率流出，如果桶滿了，溢出的上方水直接拋棄。我們把水當(dāng)作HTTP請求，每次都把請求放到一個桶中，然后以固定的速率處理請求，說了這么多，不如看一個圖加深理解(圖片來自于網(wǎng)絡(luò)，手殘黨不會畫，多多包涵)：

原理其實很簡單，就看我們怎么實現(xiàn)它了，uber團隊有一個開源的uber-go/ratelimit庫，這個庫就是漏桶的一種實現(xiàn)，下面我們一起來看一看他的實現(xiàn)思路。

樣例

學(xué)習(xí)一個新東西的時候，往往是從會用開始的，慢慢才能明白其實現(xiàn)原理，所以我們先來看看這個庫是怎樣使用的，這里我們直接提供一個實際使用例子，配合Gin框架，我們添加一個限流中間件，來達到請求限流的作用，測試代碼如下：

// 定義全局限流器對象
var rateLimit ratelimit.Limiter
// 在 gin.HandlerFunc 加入限流邏輯
func leakyBucket() gin.HandlerFunc {
	prev := time.Now()
	return func(c *gin.Context) {
		now := rateLimit.Take()
		fmt.Println(now.Sub(prev)) // 為了打印時間間隔
		prev = now // 記錄上一次的時間，沒有這個打印的會有問題
	}
}
func main() {
	rateLimit = ratelimit.New(10)
	r := gin.Default()
	r.GET("/ping", leakyBucket(), func(c *gin.Context) {
		c.JSON(200, true)
	})
	r.Run() // listen and serve on 0.0.0.0:8080 (for windows "localhost:8080")
}

我們簡單使用壓測工具ab測試一下：ab -n 10 -c 2 http://127.0.0.1:8080/ping，執(zhí)行結(jié)果部分如下：

觀察結(jié)果可知，每次處理請求的時間間隔是10ms，并且后面的請求耗時越來越久，為什么會這樣呢？ 這里先賣個小關(guān)子，看完uber的實現(xiàn)你就知道了～

源碼實現(xiàn)

我們首先來看一下其核心結(jié)構(gòu)：

type limiter struct {
	sync.Mutex
	last       time.Time
	sleepFor   time.Duration
	perRequest time.Duration
	maxSlack   time.Duration
	clock      Clock
}
type Limiter interface {
	// Take should block to make sure that the RPS is met.
	Take() time.Time
}

限制器接口只提供了一個方法take()，take()方法會阻塞確保兩次請求之間的時間走完，具體實現(xiàn)我們在下面進行分析。實現(xiàn)限制器接口的結(jié)構(gòu)體中各個字段的意義如下：

sync.Mutext：互斥鎖，控制并發(fā)的作用

last：記錄上一次的時刻

sleepFor：距離處理下一次請求需要等待的時間

perRequest：每次請求的時間間隔

maxSlack：最大松弛量，用來解決突發(fā)流量

clock：一個時鐘或模擬時鐘，提供了now和sleep方法，是實例化速率限制器

要是用該限制器，首先需要通過New方法進行初始化，一個必傳的參數(shù)是rate，代表的是每秒請求量(RPS)，還有一個可選參數(shù)，參數(shù)類型option，也就是我們可以自定義limit，不過一般使用場景不多，這里就不過多介紹了。我主要看一下他是怎么保證固定速率的，截取New方法部分代碼如下：

l := &limiter{
		perRequest: time.Second / time.Duration(rate),
		maxSlack:   -10 * time.Second / time.Duration(rate),
	}

根據(jù)我們傳入的請求數(shù)量，能計算出1s內(nèi)要通過n個請求，每個請求之間的間隔時間是多少，這樣在take方法中就可以根據(jù)這個字段來處理請求的固定速率問題，這里還初始化了最大松弛化字段，他的值是負數(shù)，默認最大松弛量是10個請求的時間間隔。

接下來我們主要看一下take方法：

func (t *limiter) Take() time.Time {
	t.Lock()
	defer t.Unlock()
	now := t.clock.Now()
	if t.last.IsZero() {
		t.last = now
		return t.last
	}
	t.sleepFor += t.perRequest - now.Sub(t.last)
	if t.sleepFor < t.maxSlack {
		t.sleepFor = t.maxSlack
	}
	if t.sleepFor > 0 {
		t.clock.Sleep(t.sleepFor)
		t.last = now.Add(t.sleepFor)
		t.sleepFor = 0
	} else {
		t.last = now
	}
	return t.last
}

take()方法的執(zhí)行步驟如下：

• 為了控制并發(fā)，所以進入該方法就需要進行上鎖，該鎖的粒度比較大，整個方法都加上了鎖
• 通過IsZero方法來判斷當(dāng)前是否是第一次請求，如果是第一次請求，直接取now時間即可返回。
• 如果不是第一次請求，就需要計算距離處理下一次請求需要等待的時間，這里有一個要注意點的是累加需要等待的時間，目的是可以給后面的抵消使用
• 如果當(dāng)前累加需要等待的時間大于最大松弛量了，將等待的時間設(shè)置為最大松弛量的時間。
• 如果當(dāng)前請求多余的時間無法完全抵消此次的所需量，調(diào)用sleep方法進行阻塞，同時清空等待的時間。如果sleepFor小于0，說明此次請求時間間隔大于預(yù)期間隔，也就說無需等待可以直接處理請求。

步驟其實不是很多，主要需要注意一個知識點 —— 最大松弛量。

漏桶算法有個天然缺陷就是無法應(yīng)對突發(fā)流量（勻速，兩次請求 req1 和 req2 之間的延遲至少應(yīng)該 >=perRequest)，舉個例子說明：假設(shè)我們現(xiàn)在有三個請求req1、req2、req3按順序處理，每個請求處理間隔為100ms，req1請求處理完成之后150ms，req2請求到來，依據(jù)限速策略可以對 req2 立即處理，當(dāng) req2 完成后，50ms 后， req3 到來，這個時候距離上次請求還不足 100ms，因此還需要等待 50ms 才能繼續(xù)執(zhí)行, 但是，對于這種情況，實際上這三個請求一共消耗了 250ms 才完成，并不是預(yù)期的 200ms。

對于上面這種情況，我們可以把之前間隔比較長的請求的時間勻給后面的請求判斷限流時使用，減少請求等待的時間了，但是當(dāng)兩個請求之間到達的間隔比較大時，就會產(chǎn)生很大的可抵消時間，以至于后面大量請求瞬間到達時，也無法抵消這個時間，那樣就已經(jīng)失去了限流的意義，所以引入了最大松弛量 (maxSlack) 的概念, 該值為負值，表示允許抵消的最長時間，防止以上情況的出現(xiàn)。

以上就是漏桶實現(xiàn)的基本思路了，整體還是很簡單的，你學(xué)會了嗎？

令牌桶算法

令牌桶其實和漏桶的原理類似，令牌桶就是想象有一個固定大小的桶，系統(tǒng)會以恒定速率向桶中放 Token，桶滿則暫時不放。從網(wǎng)上找了圖，表述非常恰當(dāng)：

關(guān)于令牌桶限流算法的實現(xiàn)，Github有一個高效的基于令牌桶限流算法實現(xiàn)的限流庫：github.com/juju/ratelimit，Golang的timer/rate也是令牌桶的一種實現(xiàn)，本文就不介紹juju/ratelimit庫了，有興趣的自己學(xué)習(xí)一下的他的實現(xiàn)思想吧，我們主要來看一看time/rate是如何實現(xiàn)的。

樣例

還是老樣子，我們還是結(jié)合gin寫一個限流中間件看看他是怎么使用的，例子如下：

import (
	"net/http"
	"time"
	"github.com/gin-gonic/gin"
	"golang.org/x/time/rate"
)
var rateLimit *rate.Limiter
func tokenBucket() gin.HandlerFunc {
	return func(c *gin.Context) {
		if rateLimit.Allow() {
			c.String(http.StatusOK, "rate limit,Drop")
			c.Abort()
			return
		}
		c.Next()
	}
}
func main() {
	limit := rate.Every(100 * time.Millisecond)
	rateLimit = rate.NewLimiter(limit, 10)
	r := gin.Default()
	r.GET("/ping", tokenBucket(), func(c *gin.Context) {
		c.JSON(200, true)
	})
	r.Run() // listen and serve on 0.0.0.0:8080 (for windows "localhost:8080")
}

上面的例子我們首先調(diào)用NewLimiter方法構(gòu)造一個限流器，第一個參數(shù)是r limit，代表每秒可以向Token桶中產(chǎn)生多少token，第二個參數(shù)是b int，代表Token桶的容量大小，對于上面的例子，表示每100ms往桶中放一個token，也就是1s鐘產(chǎn)生10個，桶的容量就是10。消費token的方法這里我們使用Allow方法，Allow 實際上就是 AllowN(time.Now(),1)，AllowN 方法表示，截止到某一時刻，目前桶中數(shù)目是否至少為 n 個，滿足則返回 true，同時從桶中消費 n 個 token。反之返回不消費 Token。對應(yīng)上面的例子，當(dāng)桶中的數(shù)目不足于1個時，就會丟掉該請求。

源碼剖析

Limit類型

time/rate自定義了一個limit類型，其實他本質(zhì)就是float64的別名，Limit定了事件的最大頻率，表示每秒事件的數(shù)據(jù)量，0就表示無限制。Inf是無限的速率限制；它允許所有事件(即使突發(fā)為0)。還提供 Every 方法來指定向 Token 桶中放置 Token 的間隔，計算出每秒時間的數(shù)據(jù)量。

type Limit float64
// Inf is the infinite rate limit; it allows all events (even if burst is zero).
const Inf = Limit(math.MaxFloat64)
// Every converts a minimum time interval between events to a Limit.
func Every(interval time.Duration) Limit {
	if interval <= 0 {
		return Inf
	}
	return 1 / Limit(interval.Seconds())
}

Limiter結(jié)構(gòu)體

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
}

各個字段含義如下：

• mu：互斥鎖、為了控制并發(fā)
• limit：每秒允許處理的事件數(shù)量，即每秒處理事件的頻率
• burst：令牌桶的最大數(shù)量，如果burst為0，并且limit == Inf，則允許處理任何事件，否則不允許
• tokens：令牌桶中可用的令牌數(shù)量
• last：記錄上次limiter的tokens被更新的時間
• lastEvent：lastEvent記錄速率受限制(桶中沒有令牌)的時間點，該時間點可能是過去的，也可能是將來的(Reservation預(yù)定的結(jié)束時間點)

Reservation結(jié)構(gòu)體

type Reservation struct {
	ok        bool
	lim       *Limiter
	tokens    int
	timeToAct time.Time
	// This is the Limit at reservation time, it can change later.
	limit Limit
}

各個字段含義如下：

ok：到截至?xí)r間是否可以獲取足夠的令牌

lim：limiter對象

tokens：需要獲取的令牌數(shù)量

timeToAct：需要等待的時間點

limit：代表預(yù)定的時間，是可以更改的。

reservation就是一個預(yù)定令牌的操作，timeToAct是本次預(yù)約需要等待到的指定時間點才有足夠預(yù)約的令牌。

Limiter消費token

Limiter有三個token的消費方法，分別是Allow、Reserve和Wait，最終三種消費方式都調(diào)用了 reserveN 、advance這兩個方法來生成和消費 Token。所以我們主要看看reserveN、advance函數(shù)的具體實現(xiàn)。

advance方法的實現(xiàn)：

func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
	//last不能在當(dāng)前時間now之后，否則計算出來的elapsed為負數(shù)，會導(dǎo)致令牌桶數(shù)量減少
  last := lim.last
	if now.Before(last) {
		last = now
	}
	//根據(jù)令牌桶的缺數(shù)計算出令牌桶未進行更新的最大時間
	maxElapsed := lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)
	elapsed := now.Sub(last) //令牌桶未進行更新的時間段
	if elapsed > maxElapsed {
		elapsed = maxElapsed
	}
	//根據(jù)未更新的時間(未向桶中加入令牌的時間段)計算出產(chǎn)生的令牌數(shù)
	delta := lim.limit.tokensFromDuration(elapsed)
	tokens := lim.tokens + delta //計算出可用的令牌數(shù)
	if burst := float64(lim.burst); tokens > burst {
		tokens = burst
	}
	return now, last, tokens
}

advance方法的作用是更新令牌桶的狀態(tài)，計算出令牌桶未更新的時間(elapsed)，根據(jù)elapsed算出需要向桶中加入的令牌數(shù)delta，然后算出桶中可用的令牌數(shù)newTokens.

reserveN方法的實現(xiàn)：reserveN是 AllowN, ReserveN及 WaitN的輔助方法，用于判斷在maxFutureReserve時間內(nèi)是否有足夠的令牌。

// @param n 要消費的token數(shù)量
// @param maxFutureReserve 愿意等待的最長時間
func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation {
	lim.mu.Lock()
	// 如果沒有限制
	if lim.limit == Inf {
		lim.mu.Unlock()
		return Reservation{
			ok:        true, //桶中有足夠的令牌
			lim:       lim,
			tokens:    n,
			timeToAct: now,
		}
	}
	//更新令牌桶的狀態(tài)，tokens為目前可用的令牌數(shù)量
	now, last, tokens := lim.advance(now)
  // 計算取完之后桶還能剩能下多少token
	tokens -= float64(n)
	var waitDuration time.Duration
  // 如果token < 0, 說明目前的token不夠，需要等待一段時間
	if tokens < 0 {
		waitDuration = lim.limit.durationFromTokens(-tokens)
	}
	ok := n <= lim.burst && waitDuration <= maxFutureReserve
	r := Reservation{
		ok:    ok,
		lim:   lim,
		limit: lim.limit,
	}
  // timeToAct表示當(dāng)桶中滿足token數(shù)目等于n的時間
	if ok {
		r.tokens = n
		r.timeToAct = now.Add(waitDuration)
	}
  // 更新桶里面的token數(shù)目
	// 更新last時間
	// lastEvent
	if ok {
		lim.last = now
		lim.tokens = tokens
		lim.lastEvent = r.timeToAct
	} else {
		lim.last = last
	}
	lim.mu.Unlock()
	return r
}

上面的代碼我已經(jīng)進行了注釋，這里在總結(jié)一下流程：

• 首選判斷是否擁有速率限制，沒有速率限制也就是桶中一致?lián)碛凶銐虻牧钆啤?/li>
• 計算從上次取 Token 的時間到當(dāng)前時刻，期間一共新產(chǎn)生了多少 Token：我們只在取 Token 之前生成新的 Token，也就意味著每次取Token的間隔，實際上也是生成 Token 的間隔。我們可以利用 tokensFromDuration, 輕易的算出這段時間一共產(chǎn)生 Token 的數(shù)目。所以當(dāng)前 Token 數(shù)目 = 新產(chǎn)生的 Token 數(shù)目 + 之前剩余的 Token 數(shù)目 - 要消費的 Token 數(shù)目。
• 如果消費后剩余 Token 數(shù)目大于零，說明此時 Token 桶內(nèi)仍不為空，此時 Token 充足，無需調(diào)用側(cè)等待。 如果 Token 數(shù)目小于零，則需等待一段時間。那么這個時候，我們可以利用 durationFromTokens 將當(dāng)前負值的 Token 數(shù)轉(zhuǎn)化為需要等待的時間。
• 將需要等待的時間等相關(guān)結(jié)果返回給調(diào)用方

其實整個過程就是利用了 Token 數(shù)可以和時間相互轉(zhuǎn)化 的原理。而如果 Token 數(shù)為負，則需要等待相應(yīng)時間即可。

上面提到了durationFromTokens、tokensFromDuration這兩個方法，是關(guān)鍵，他們的實現(xiàn)如下：

func (limit Limit) durationFromTokens(tokens float64) time.Duration {
	seconds := tokens / float64(limit)
	return time.Nanosecond * time.Duration(1e9*seconds)
}
func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	// Split the integer and fractional parts ourself to minimize rounding errors.
	// See golang.org/issues/34861.
	sec := float64(d/time.Second) * float64(limit)
	nsec := float64(d%time.Second) * float64(limit)
	return sec + nsec/1e9
}

• durationFromTokens：功能是計算出生成N 個新的 Token 一共需要多久。
• tokensFromDuration：給定一段時長，這段時間一共可以生成多少個 Token。

細心的網(wǎng)友會發(fā)現(xiàn)tokensFromDuration方法既然是計算一段時間一共可以生成多少個 Token，為什么不直接進行相乘呢？其實Golang最初的版本就是采用d.Seconds() * float64(limit)直接相乘實現(xiàn)的，雖然看上去一點問題沒有，但是這里是兩個小數(shù)相乘，會帶來精度損失，所以采用現(xiàn)在這種方法實現(xiàn)，分別求出秒的整數(shù)部分和小數(shù)部分，進行相乘后再相加，這樣可以得到最精確的精度。

limiter歸還Token

既然我們可以消費Token，那么對應(yīng)也可以取消此次消費，將token歸還，當(dāng)調(diào)用 Cancel() 函數(shù)時，消費的 Token 數(shù)將會盡可能歸還給 Token 桶。歸還也并不是那么簡單，接下我們我們看看歸還token是如何實現(xiàn)的。

func (r *Reservation) CancelAt(now time.Time) {
	if !r.ok {
		return
	}
	r.lim.mu.Lock()
	defer r.lim.mu.Unlock()
  /*
  1.如果無需限流
	2. tokens為0 (需要獲取的令牌數(shù)量為0)
	3. 已經(jīng)過了截至?xí)r間
	以上三種情況無需處理取消操作
	*/
	if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(now) {
		return
	}
	//計算出需要還原的令牌數(shù)量
	//這里的r.lim.lastEvent可能是本次Reservation的結(jié)束時間，也可能是后來的Reservation的結(jié)束時間，所以要把本次結(jié)束時間點(r.timeToAct)之后產(chǎn)生的令牌數(shù)減去
	restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))
  // 當(dāng)小于0，表示已經(jīng)都預(yù)支完了，不能歸還了
	if restoreTokens <= 0 {
		return
	}
	//從新計算令牌桶的狀態(tài)
	now, _, tokens := r.lim.advance(now)
	//還原當(dāng)前令牌桶的令牌數(shù)量，當(dāng)前的令牌數(shù)tokens加上需要還原的令牌數(shù)restoreTokens
	tokens += restoreTokens
  //如果tokens大于桶的最大容量，則將tokens置為桶的最大容量
	if burst := float64(r.lim.burst); tokens > burst {
		tokens = burst
	}
	// update state
	r.lim.last = now //記錄桶的更新時間
	r.lim.tokens = tokens //更新令牌數(shù)量
 // 如果都相等，說明跟沒消費一樣。直接還原成上次的狀態(tài)吧
	if r.timeToAct == r.lim.lastEvent {
		prevEvent := r.timeToAct.Add(r.limit.durationFromTokens(float64(-r.tokens)))
		if !prevEvent.Before(now) {
			r.lim.lastEvent = prevEvent
		}
	}
	return
}

注釋已經(jīng)添加，就不在詳細解釋了，重點是這一行代碼：

restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))，

• r.tokens指的是本次消費的token數(shù)，
• r.timeToAcr指的是Token桶可以滿足本次消費數(shù)目的時刻，也就是消費的時刻+等待的時長
• r.lim.lastEvent指的是最近一次消費的timeToAct的值，

通過r.limit.tokensFromDuration方法得出的結(jié)果指的是從該次消費到當(dāng)前時間，一共又消費了多少Token數(shù)目，所以最終得出這一段的代碼含義是：

要歸還的Token = 該次消費的Token - 新消費的token。

好啦，源碼就暫時分析到這了，因為標準庫的實現(xiàn)的代碼量有點大，還有一部分在這里沒有說，留給大家自己去剖析吧～。

總結(jié)

本文重點介紹了漏桶算法和令牌桶算法，漏桶算法和令牌桶算法的主要區(qū)別在于，"漏桶算法"能夠強行限制數(shù)據(jù)的傳輸速率(或請求頻率)，而"令牌桶算法"在能夠限制數(shù)據(jù)的平均傳輸速率外，還允許某種程度的突發(fā)傳輸。在某些情況下，漏桶算法不能夠有效地使用網(wǎng)絡(luò)資源，因為漏桶的漏出速率是固定的，所以即使網(wǎng)絡(luò)中沒有發(fā)生擁塞，漏桶算法也不能使某一個單獨的數(shù)據(jù)流達到端口速率。因此，漏桶算法對于存在突發(fā)特性的流量來說缺乏效率。而令牌桶算法則能夠滿足這些具有突發(fā)特性的流量。通常，漏桶算法與令牌桶算法結(jié)合起來為網(wǎng)絡(luò)流量提供更高效的控制。

文中測試代碼已上傳：

https://github.com/asong2020/Golang_Dream/tree/master/code_demo/limit_demo

更多關(guān)于golang高并發(fā)限流漏桶令牌桶的資料請關(guān)注腳本之家其它相關(guān)文章！