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

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

前言

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

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

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

1.問題描述

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

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

2.信號量限流

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

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

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

2.1阻塞方式

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

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

typeSemaphorestruct{

innerChanchanstruct{}

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

funcNewSemaphore(numuint64)*Semaphore{

returnSemaphore{

innerChan:make(chanstruct{},num),

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

func(s*Semaphore)Acquire(){

for{

select{

cases.innerChan-struct{}{}:

return

default:

log.Error("semaphoreacquireisblocking")

time.Sleep(100*time.Millisecond)

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

func(s*Semaphore)Release(){

select{

case-s.innerChan:

return

default:

return

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

2.2非阻塞方式

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

3.限流算法

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

3.1漏桶算法

漏桶(LeakyBucket)算法思路很簡單，水(請求)先進(jìn)入到漏桶里，漏桶以一定的速度出水(接口有響應(yīng)速率)，當(dāng)水流入速度過大會直接溢出(訪問頻率超過接口響應(yīng)速率)然后就拒絕請求。可以看出漏桶算法能強行限制數(shù)據(jù)的傳輸速率。示意圖如下：

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

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

3.2令牌桶算法

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

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

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

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

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

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

func(sp*servicePanel)incLimit()error{

//如果大于限制的條件則返回錯誤

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

returnErrCurrentLimit

sp.currentLimitCount.Inc()

returnnil

func(sp*servicePanel)clearLimit(){

//定期每秒重置計數(shù)器，從而達(dá)到每秒限制的并發(fā)數(shù)

//比如限制1000req/s，在這里指每秒清理1000的計數(shù)值

//令牌桶是定期放，這里是逆思維，每秒清空，實現(xiàn)不僅占用內(nèi)存低而且效率高

t:=time.NewTicker(time.Second)

for{

select{

case-t.C:

sp.currentLimitCount.Store(0)

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

改進(jìn)

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

func(sp*servicePanel)incLimit()error{

//如果大于1則返回錯誤

ifsp.currentLimitCount.Load()1{

returnErrCurrentLimit

sp.currentLimitCount.Inc()

returnnil

func(sp*servicePanel)clearLimit(){

//1s除以每秒限流個數(shù)

t:=time.NewTicker(time.Second/time.Duration(sp.currLimitFunc(nil)))

for{

select{

case-t.C:

sp.currentLimitCount.Store(0)

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

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

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

4.1引入方式

#第一版本

goget/uber-go/ratelimit@v0.1.0

#改進(jìn)版本

goget/uber-go/ratelimit@master

4.2使用

首先強調(diào)一點，跟筆者自研的限流器最大的不同的是，這是一個阻塞調(diào)用者的限流組件。限流速率一般表示為rate/s即一秒內(nèi)rate個請求。先不多說，進(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

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

構(gòu)造限流器

首先是構(gòu)造一個Limiter里面有一個perRequest這是關(guān)鍵的一個變量，表示每個請求之間相差的間隔時間，這是此組件的算法核心思想，也就是說將請求排隊，一秒之內(nèi)有rate個請求，將這些請求排隊，挨個來，每個請求的間隔就是1s/rate從來達(dá)到1s內(nèi)rate個請求的概念，從而達(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()方法每次請求前使用，用來獲取批準(zhǔn)返回批準(zhǔn)時刻的時間。

第一版本

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

在實現(xiàn)方面，可以看到第一版本采用了Go的lock，然后排隊sleep，完成sleep之后，請求之間的間隔時間恒定，單位時間之內(nèi)有設(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