前言

限流器，顧名思義用來對高并發(fā)的請求進(jìn)行流量限制的組件。

限流包括 Nginx 層面的限流以及業(yè)務(wù)代碼邏輯上的限流。流量的限制在眾多微服務(wù)和 service mesh 中多有應(yīng)用。限流主要有三種算法：信號量、漏桶算法和令牌桶算法。下面依次介紹這三種算法。

筆者在本文的程序示例均以 Go 語言實(shí)現(xiàn)。

1. 問題描述

用戶增長過快、熱門業(yè)務(wù)或者爬蟲等惡意攻擊行為致使請求量突然增大，比如學(xué)校的教務(wù)系統(tǒng)，到了查分之日，請求量漲到之前的 100 倍都不止，沒多久該接口幾乎不可使用，并引發(fā)連鎖反應(yīng)導(dǎo)致整個(gè)系統(tǒng)崩潰。如何應(yīng)對這種情況呢？生活給了我們答案：比如老式電閘都安裝了保險(xiǎn)絲，一旦有人使用超大功率的設(shè)備，保險(xiǎn)絲就會燒斷以保護(hù)各個(gè)電器不被強(qiáng)電流給燒壞。同理我們的接口也需要安裝上“保險(xiǎn)絲”，以防止非預(yù)期的請求對系統(tǒng)壓力過大而引起的系統(tǒng)癱瘓，當(dāng)流量過大時(shí)，可以采取拒絕或者引流等機(jī)制。

后端服務(wù)由于各個(gè)業(yè)務(wù)的不同和復(fù)雜性，各自在容器部署的時(shí)候都可能會有單臺的瓶頸，超過瓶頸會導(dǎo)致內(nèi)存或者 cpu 的瓶頸，進(jìn)而導(dǎo)致發(fā)生服務(wù)不可用或者單臺容器直接掛掉或重啟。

2. 信號量限流

信號量在眾多開發(fā)語言中都會有相關(guān)信號量的設(shè)計(jì)。如 Java 中的Semaphore 是一個(gè)計(jì)數(shù)信號量。常用于限制獲取某資源的線程數(shù)量，可基于 Java 的 concurrent 并發(fā)包實(shí)現(xiàn)。

信號量兩個(gè)重要方法 Acquire() 和 Release()。通過acquire()方法獲取許可，該方法會阻塞，直到獲取許可為止。通過release()方法釋放許可。

筆者在閱讀一些語言開源實(shí)現(xiàn)后，總結(jié)出信號量的主要有非阻塞和阻塞兩種。

2.1 阻塞方式

采用鎖或者阻塞隊(duì)列方式，以 Go 語言為示例如下：

// 采用channel作為底層數(shù)據(jù)結(jié)構(gòu)，從而達(dá)到阻塞的獲取和使用信號量
type Semaphore struct {
	innerChan chan struct{}
}
// 初始化信號量，本質(zhì)初始化一個(gè)channel，channel的初始化大小為 信號量數(shù)值
func NewSemaphore(num uint64) *Semaphore {
	return &Semaphore{
		innerChan: make(chan struct{}, num),
	}
}
// 獲取信號量，本質(zhì)是 向channel放入元素，如果同時(shí)有很多協(xié)程并發(fā)獲取信號量，則channel則會full阻塞，從而達(dá)到控制并發(fā)協(xié)程數(shù)的目的，也即是信號量的控制
func (s *Semaphore) Acquire() {
	for {
		select {
		case s.innerChan <- struct{}{}:
			return
		default:
			log.Error("semaphore acquire is blocking")
			time.Sleep(100 * time.Millisecond)
		}
	}
}
// 釋放信號量 本質(zhì)是 從channel中獲取元素，由于有acquire的放入元素，所以此處一定能回去到元素 也就能釋放成功，default只要是出于安全編程的目的
func (s *Semaphore) Release() {
	select {
	case <-s.innerChan:
		return
	default:
		return
	}
}

在實(shí)現(xiàn)中，定義了 Semaphore 結(jié)構(gòu)體。初始化信號量，本質(zhì)是初始化一個(gè)channel，channel 的初始化大小為信號量數(shù)值；獲取信號量，本質(zhì)是向channel放入元素，如果同時(shí)有很多協(xié)程并發(fā)獲取信號量，則 channel 則會 full 阻塞，從而達(dá)到控制并發(fā)協(xié)程數(shù)的目的，也即是信號量的控制；釋放信號量的本質(zhì)是從channel中獲取元素，由于有acquire的放入元素，所以此處一定能回去到元素 也就能釋放成功，default只要是出于安全編程的目的。

2.2 非阻塞方式

以并發(fā)安全的計(jì)數(shù)方式比如采用原子 atomic 加減進(jìn)行。

3. 限流算法

主流的限流算法分為兩種漏桶算法和令牌桶算法，關(guān)于這兩個(gè)算法有很多文章和論文都給出了詳細(xì)的講解。從原理上看，令牌桶算法和漏桶算法是相反的，一個(gè) 進(jìn)水，一個(gè)是 漏水。值得一提的是 Google Guava 開源和 Uber 開源限流組件均采用漏桶算法。

3.1 漏桶算法

漏桶(Leaky Bucket)算法思路很簡單，水(請求)先進(jìn)入到漏桶里，漏桶以一定的速度出水(接口有響應(yīng)速率)，當(dāng)水流入速度過大會直接溢出(訪問頻率超過接口響應(yīng)速率)然后就拒絕請求?？梢钥闯雎┩八惴軓?qiáng)行限制數(shù)據(jù)的傳輸速率。示意圖如下：

可見這里有兩個(gè)變量，一個(gè)是桶的大小，支持流量突發(fā)增多時(shí)可以存多少的水(burst)，另一個(gè)是水桶漏洞的大小(rate)。

漏桶算法可以使用 redis 隊(duì)列來實(shí)現(xiàn)，生產(chǎn)者發(fā)送消息前先檢查隊(duì)列長度是否超過閾值，超過閾值則丟棄消息，否則發(fā)送消息到 Redis 隊(duì)列中；消費(fèi)者以固定速率從 Redis 隊(duì)列中取消息。Redis 隊(duì)列在這里起到了一個(gè)緩沖池的作用，起到削峰填谷、流量整形的作用。

3.2 令牌桶算法

對于很多應(yīng)用場景來說，除了要求能夠限制數(shù)據(jù)的平均傳輸速率外，還要求允許某種程度的突發(fā)傳輸。這時(shí)候漏桶算法可能就不合適了，令牌桶算法更為適合。令牌桶算法的原理是系統(tǒng)會以一個(gè)恒定的速度往桶里放入令牌，而如果請求需要被處理，則需要先從桶里獲取一個(gè)令牌，當(dāng)桶里沒有令牌可取時(shí)，則拒絕服務(wù)。桶里能夠存放令牌的最高數(shù)量，就是允許的突發(fā)傳輸量。

放令牌這個(gè)動作是持續(xù)不斷的進(jìn)行，如果桶中令牌數(shù)達(dá)到上限，就丟棄令牌，所以就存在這種情況，桶中一直有大量的可用令牌，這時(shí)進(jìn)來的請求就可以直接拿到令牌執(zhí)行，比如設(shè)置qps為100，那么限流器初始化完成一秒后，桶中就已經(jīng)有100個(gè)令牌了，等啟動完成對外提供服務(wù)時(shí)，該限流器可以抵擋瞬時(shí)的100個(gè)請求。所以，只有桶中沒有令牌時(shí)，請求才會進(jìn)行等待，最后相當(dāng)于以一定的速率執(zhí)行。

可以準(zhǔn)備一個(gè)隊(duì)列，用來保存令牌，另外通過一個(gè)線程池定期生成令牌放到隊(duì)列中，每來一個(gè)請求，就從隊(duì)列中獲取一個(gè)令牌，并繼續(xù)執(zhí)行。

3.3 漏桶算法的實(shí)現(xiàn)

所以此處筆者開門見山，直接展示此算法的 Go 語言版本的實(shí)現(xiàn)，代碼如下：

// 此處截取自研的熔斷器代碼中的限流實(shí)現(xiàn)，這是非阻塞的實(shí)現(xiàn)
func (sp *servicePanel) incLimit() error {
	// 如果大于限制的條件則返回錯(cuò)誤
	if sp.currentLimitCount.Load() > sp.currLimitFunc(nil) {
		return ErrCurrentLimit
	}
	sp.currentLimitCount.Inc()
	return nil
}
func (sp *servicePanel) clearLimit() {
	// 定期每秒重置計(jì)數(shù)器，從而達(dá)到每秒限制的并發(fā)數(shù)
	// 比如限制1000req/s，在這里指每秒清理1000的計(jì)數(shù)值
// 令牌桶是定期放，這里是逆思維，每秒清空，實(shí)現(xiàn)不僅占用內(nèi)存低而且效率高
	t := time.NewTicker(time.Second)
	for {
		select {
		case <-t.C:
			sp.currentLimitCount.Store(0)
		}
	}
}

上述的實(shí)現(xiàn)實(shí)際是比較粗糙的實(shí)現(xiàn)，沒有嚴(yán)格按照每個(gè)請求方按照某個(gè)固定速率進(jìn)行，而是以秒為單位，粗粒度的進(jìn)行計(jì)數(shù)清零，這其實(shí)會造成某個(gè)瞬間雙倍的每秒限流個(gè)數(shù)，雖然看上去不滿足要求，但是在這個(gè)瞬間其實(shí)是只是一個(gè)雙倍值，正常系統(tǒng)都應(yīng)該會應(yīng)付一瞬間雙倍限流個(gè)數(shù)的請求量。

改進(jìn)

如果要嚴(yán)格的按照每個(gè)請求按照某個(gè)固定數(shù)值進(jìn)行，那么可以改進(jìn)時(shí)間的粗力度，具體做法如下：

func (sp *servicePanel) incLimit() error {
	// 如果大于1則返回錯(cuò)誤
	if sp.currentLimitCount.Load() > 1 {
		return ErrCurrentLimit
	}
	sp.currentLimitCount.Inc()
	return nil
}
func (sp *servicePanel) clearLimit() {
	// 1s除以每秒限流個(gè)數(shù)
	t := time.NewTicker(time.Second/time.Duration(sp.currLimitFunc(nil)))
	for {
		select {
		case <-t.C:
			sp.currentLimitCount.Store(0)
		}
	}
}

讀者可以自行嘗試一下改進(jìn)之后的漏斗算法。

4. Uber 開源實(shí)現(xiàn) RateLimit 深入解析

uber 在 Github 上開源了一套用于服務(wù)限流的 go 語言庫 ratelimit, 該組件基于 Leaky Bucket(漏桶)實(shí)現(xiàn)。

4.1 引入方式

#第一版本
go get github.com/uber-go/ratelimit@v0.1.0
#改進(jìn)版本
go get github.com/uber-go/ratelimit@master

4.2 使用

首先強(qiáng)調(diào)一點(diǎn)，跟筆者自研的限流器最大的不同的是，這是一個(gè)阻塞調(diào)用者的限流組件。限流速率一般表示為 rate/s 即一秒內(nèi) rate 個(gè)請求。先不多說，進(jìn)行一下用法示例：

func ExampleRatelimit() {
	rl := ratelimit.New(100) // per second
	prev := time.Now()
	for i := 0; i < 10; i++ {
		now := rl.Take()
		if i > 0 {
			fmt.Println(i, now.Sub(prev))
		}
		prev = now
	}
}

預(yù)期的結(jié)果如下：

    // Output:
    // 1 10ms
    // 2 10ms
    // 3 10ms
    // 4 10ms
    // 5 10ms
    // 6 10ms
    // 7 10ms
    // 8 10ms
    // 9 10ms

測試結(jié)果完全符合預(yù)期。在這個(gè)例子中，我們給定限流器每秒可以通過100個(gè)請求，也就是平均每個(gè)請求間隔10ms。因此，最終會每10ms打印一行數(shù)據(jù)。

構(gòu)造限流器

首先是構(gòu)造一個(gè)Limiter 里面有一個(gè) perRequest 這是關(guān)鍵的一個(gè)變量，表示每個(gè)請求之間相差的間隔時(shí)間，這是此組件的算法核心思想，也就是說將請求排隊(duì)，一秒之內(nèi)有rate個(gè)請求，將這些請求排隊(duì)，挨個(gè)來，每個(gè)請求的間隔就是1s/rate 從來達(dá)到 1s內(nèi)rate個(gè)請求的概念，從而達(dá)到限流的目的。

// New returns a Limiter that will limit to the given RPS.
func New(rate int, opts ...Option) Limiter {
	l := &limiter{
		perRequest: time.Second / time.Duration(rate),
		maxSlack:   -10 * time.Second / time.Duration(rate),
	}
	for _, opt := range opts {
		opt(l)
	}
	if l.clock == nil {
		l.clock = clock.New()
	}
	return l
}

限流器Take() 阻塞方法

Take() 方法 每次請求前使用，用來獲取批準(zhǔn) 返回批準(zhǔn)時(shí)刻的時(shí)間。

第一版本

// Take blocks to ensure that the time spent between multiple
// Take calls is on average time.Second/rate.
func (t *limiter) Take() time.Time {
	t.Lock()
	defer t.Unlock()
	now := t.clock.Now()
	// If this is our first request, then we allow it.
	if t.last.IsZero() {
		t.last = now
		return t.last
	}
	// sleepFor calculates how much time we should sleep based on
	// the perRequest budget and how long the last request took.
	// Since the request may take longer than the budget, this number
	// can get negative, and is summed across requests.
	t.sleepFor += t.perRequest - now.Sub(t.last)
	// We shouldn't allow sleepFor to get too negative, since it would mean that
	// a service that slowed down a lot for a short period of time would get
	// a much higher RPS following that.
	if t.sleepFor < t.maxSlack {
		t.sleepFor = t.maxSlack
	}
	// If sleepFor is positive, then we should sleep now.
	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
}

在實(shí)現(xiàn)方面，可以看到第一版本采用了 Go 的 lock，然后排隊(duì) sleep，完成 sleep 之后，請求之間的間隔時(shí)間恒定，單位時(shí)間之內(nèi)有設(shè)定好的請求數(shù)，實(shí)現(xiàn)限流的目的。

第二版本

// Take blocks to ensure that the time spent between multiple
// Take calls is on average time.Second/rate.
func (t *limiter) Take() time.Time {
	newState := state{}
	taken := false
	for !taken {
		now := t.clock.Now()
		previousStatePointer := atomic.LoadPointer(&t.state)
		oldState := (*state)(previousStatePointer)
		newState = state{}
		newState.last = now
		// If this is our first request, then we allow it.
		if oldState.last.IsZero() {
			taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
			continue
		}
		// sleepFor calculates how much time we should sleep based on
		// the perRequest budget and how long the last request took.
		// Since the request may take longer than the budget, this number
		// can get negative, and is summed across requests.
		newState.sleepFor += t.perRequest - now.Sub(oldState.last)
		// We shouldn't allow sleepFor to get too negative, since it would mean that
		// a service that slowed down a lot for a short period of time would get
		// a much higher RPS following that.
		if newState.sleepFor < t.maxSlack {
			newState.sleepFor = t.maxSlack
		}
		if newState.sleepFor > 0 {
			newState.last = newState.last.Add(newState.sleepFor)
		}
		taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
	}
	t.clock.Sleep(newState.sleepFor)
	return newState.last
}

第二版本采用原子操作+for的自旋操作來替代lock操作，這樣做的目的是減少協(xié)程鎖競爭。 兩個(gè)版本不管是用鎖還是原子操作本質(zhì)都是讓請求排隊(duì)，第一版本存在鎖競爭，然后排隊(duì)sleep，第二版本避免鎖競爭，但是所有協(xié)程可能很快跳出for循環(huán)然后都會在sleep處sleep。

小結(jié)

保障服務(wù)穩(wěn)定的三大利器：熔斷降級、服務(wù)限流和故障模擬。本文主要講解了分布式系統(tǒng)中高可用的常用策略：限流。限流通常有三種實(shí)現(xiàn)：信號量（計(jì)數(shù)器）、漏桶、令牌桶。本文基于漏桶算法實(shí)現(xiàn)了一個(gè)限流小插件。最后分析了 uber 開源的 uber-go，限流器 Take() 阻塞方法的第二版本對協(xié)程鎖競爭更加友好。

參考 http://www.dhdzp.com/article/251947.htm

以上就是Go實(shí)現(xiàn)分布式系統(tǒng)高可用限流器實(shí)戰(zhàn)的詳細(xì)內(nèi)容，更多關(guān)于Go分布式系統(tǒng)高可用限流器的資料請關(guān)注腳本之家其它相關(guān)文章！