Go實(shí)現(xiàn)分布式系統(tǒng)高可用限流器實(shí)戰(zhàn)

第Go實(shí)現(xiàn)分布式系統(tǒng)高可用限流器實(shí)戰(zhàn)目錄前言1.問題描述2.信號(hào)量限流2.1阻塞方式2.2非阻塞方式3.限流算法3.1漏桶算法3.2令牌桶算法3.3漏桶算法的實(shí)現(xiàn)改進(jìn)4.Uber開源實(shí)現(xiàn)RateLimit深入解析4.1引入方式4.2使用構(gòu)造限流器限流器Take()阻塞方法第一版本第二版本小結(jié)

前言

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

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

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

1.問題描述

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

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

2.信號(hào)量限流

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

信號(hào)量?jī)蓚€(gè)重要方法Acquire()和Release()。通過acquire()方法獲取許可，該方法會(huì)阻塞，直到獲取許可為止。通過release()方法釋放許可。

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

2.1阻塞方式

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

//采用channel作為底層數(shù)據(jù)結(jié)構(gòu)，從而達(dá)到阻塞的獲取和使用信號(hào)量

typeSemaphorestruct{

innerChanchanstruct{}

//初始化信號(hào)量，本質(zhì)初始化一個(gè)channel，channel的初始化大小為信號(hào)量數(shù)值

funcNewSemaphore(numuint64)*Semaphore{

returnSemaphore{

innerChan:make(chanstruct{},num),

//獲取信號(hào)量，本質(zhì)是向channel放入元素，如果同時(shí)有很多協(xié)程并發(fā)獲取信號(hào)量，則channel則會(huì)full阻塞，從而達(dá)到控制并發(fā)協(xié)程數(shù)的目的，也即是信號(hào)量的控制

func(s*Semaphore)Acquire(){

for{

select{

cases.innerChan-struct{}{}:

return

default:

log.Error("semaphoreacquireisblocking")

time.Sleep(100*time.Millisecond)

//釋放信號(hào)量本質(zhì)是從channel中獲取元素，由于有acquire的放入元素，所以此處一定能回去到元素也就能釋放成功，default只要是出于安全編程的目的

func(s*Semaphore)Release(){

select{

case-s.innerChan:

return

default:

return

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

2.2非阻塞方式

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

3.限流算法

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

3.1漏桶算法

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

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

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

3.2令牌桶算法

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

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

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

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

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

//此處截取自研的熔斷器代碼中的限流實(shí)現(xiàn)，這是非阻塞的實(shí)現(xiàn)

func(sp*servicePanel)incLimit()error{

//如果大于限制的條件則返回錯(cuò)誤

ifsp.currentLimitCount.Load()sp.currLimitFunc(nil){

returnErrCurrentLimit

sp.currentLimitCount.Inc()

returnnil

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è)請(qǐng)求方按照某個(gè)固定速率進(jìn)行，而是以秒為單位，粗粒度的進(jìn)行計(jì)數(shù)清零，這其實(shí)會(huì)造成某個(gè)瞬間雙倍的每秒限流個(gè)數(shù)，雖然看上去不滿足要求，但是在這個(gè)瞬間其實(shí)是只是一個(gè)雙倍值，正常系統(tǒng)都應(yīng)該會(huì)應(yīng)付一瞬間雙倍限流個(gè)數(shù)的請(qǐng)求量。

改進(jìn)

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

func(sp*servicePanel)incLimit()error{

//如果大于1則返回錯(cuò)誤

ifsp.currentLimitCount.Load()1{

returnErrCurrentLimit

sp.currentLimitCount.Inc()

returnnil

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語(yǔ)言庫(kù)ratelimit,該組件基于LeakyBucket(漏桶)實(shí)現(xiàn)。

4.1引入方式

#第一版本

goget/uber-go/ratelimit@v0.1.0

#改進(jìn)版本

goget/uber-go/ratelimit@master

4.2使用

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

funcExampleRatelimit(){

rl:=ratelimit.New(100)//persecond

prev:=time.Now()

fori:=0;ii++{

now:=rl.Take()

ifi0{

fmt.Println(i,now.Sub(prev))

prev=now

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

//Output:

//110ms

//210ms

//310ms

//410ms

//510ms

//610ms

//710ms

//810ms

//910ms

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

構(gòu)造限流器

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

//NewreturnsaLimiterthatwilllimittothegivenRPS.

funcNew(rateint,opts...Option)Limiter{

l:=limiter{

perRequest:time.Second/time.Duration(rate),

maxSlack:-10*time.Second/time.Duration(rate),

for_,opt:=rangeopts{

opt(l)

ifl.clock==nil{

l.clock=clock.New()

returnl

限流器Take()阻塞方法

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

第一版本

//Takeblockstoensurethatthetimespentbetweenmultiple

//Takecallsisonaveragetime.Second/rate.

func(t*limiter)Take()time.Time{

t.Lock()

defert.Unlock()

now:=t.clock.Now()

//Ifthisisourfirstrequest,thenweallowit.

ift.last.IsZero(){

t.last=now

returnt.last

//sleepForcalculateshowmuchtimeweshouldsleepbasedon

//theperRequestbudgetandhowlongthelastrequesttook.

//Sincetherequestmaytakelongerthanthebudget,thisnumber

//cangetnegative,andissummedacrossrequests.

t.sleepFor+=t.perRequest-now.Sub(t.last)

//Weshouldn'tallowsleepFortogettoonegative,sinceitwouldmeanthat

//aservicethatsloweddownalotforashortperiodoftimewouldget

//amuchhigherRPSfollowingthat.

ift.sleepFort.maxSlack{

t.sleepFor=t.maxSlack

//IfsleepForispositive,thenweshouldsleepnow.

ift.sleepFor0{

t.clock.Sleep(t.sleepFor)

t.last=now.Add(t.sleepFor)

t.sleepFor=0

}else{

t.last=now

returnt.last

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

第二版本

//Takeblockstoensurethatthetimespentbetweenmultiple

//Takecallsisonaveragetime.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

//Ifthisisourfirstrequest,thenweallowit.

ifoldState.last.IsZero(){

taken=atomic.CompareAndSwapPointer(t.state,previousStatePointer,unsafe.Pointer(newState))

continue

//sleepForcalculateshowmuchtimeweshouldsleepbasedon

//theperRequestbudgetandhowlongthelastrequesttook.

//Sincetherequestmaytakelongerthanthebudget,thisnumber

//cangetnegative,andissummedacrossrequests.

newState.sleepFor+=t.perRequest-now.Sub(oldState.last)

//Weshouldn'tallowsleepFortogettoonegative,sinceitwouldmeanthat