go熔斷原理分析與源碼解讀

第go熔斷原理分析與源碼解讀目錄正文熔斷原理熔斷器實(shí)現(xiàn)hystrixBreaker和googlebreaker對比源碼解讀結(jié)束語

正文

熔斷機(jī)制（CircuitBreaker）指的是在股票市場的交易時(shí)間中，當(dāng)價(jià)格的波動(dòng)幅度達(dá)到某一個(gè)限定的目標(biāo)（熔斷點(diǎn)）時(shí)，對其暫停交易一段時(shí)間的機(jī)制。此機(jī)制如同保險(xiǎn)絲在電流過大時(shí)候熔斷，故而得名。熔斷機(jī)制推出的目的是為了防范系統(tǒng)性風(fēng)險(xiǎn)，給市場更多的冷靜時(shí)間，避免恐慌情緒蔓延導(dǎo)致整個(gè)市場波動(dòng)，從而防止大規(guī)模股價(jià)下跌現(xiàn)象的發(fā)生。

同樣的，在高并發(fā)的分布式系統(tǒng)設(shè)計(jì)中，也應(yīng)該有熔斷的機(jī)制。熔斷一般是在客戶端（調(diào)用端）進(jìn)行配置，當(dāng)客戶端向服務(wù)端發(fā)起請求的時(shí)候，服務(wù)端的錯(cuò)誤不斷地增多，這時(shí)候就可能會(huì)觸發(fā)熔斷，觸發(fā)熔斷后客戶端的請求不再發(fā)往服務(wù)端，而是在客戶端直接拒絕請求，從而可以保護(hù)服務(wù)端不會(huì)過載。這里說的服務(wù)端可能是rpc服務(wù)，http服務(wù)，也可能是mysql，redis等。注意熔斷是一種有損的機(jī)制，當(dāng)熔斷后可能需要一些降級(jí)的策略進(jìn)行配合。

熔斷原理

現(xiàn)代微服務(wù)架構(gòu)基本都是分布式的，整個(gè)分布式系統(tǒng)是由非常多的微服務(wù)組成。不同服務(wù)之間相互調(diào)用，組成復(fù)雜的調(diào)用鏈路。在復(fù)雜的調(diào)用鏈路中的某一個(gè)服務(wù)如果不穩(wěn)定，就可能會(huì)層層級(jí)聯(lián)，最終可能導(dǎo)致整個(gè)鏈路全部掛掉。因此我們需要對不穩(wěn)定的服務(wù)依賴進(jìn)行熔斷降級(jí)，暫時(shí)切斷不穩(wěn)定的服務(wù)調(diào)用，避免局部不穩(wěn)定因素導(dǎo)致整個(gè)分布式系統(tǒng)的雪崩。

說白了，我覺得熔斷就像是那些容易異常服務(wù)的一種代理，這個(gè)代理能夠記錄最近調(diào)用發(fā)生錯(cuò)誤的次數(shù)，然后決定是繼續(xù)操作，還是立即返回錯(cuò)誤。

熔斷器內(nèi)部維護(hù)了一個(gè)熔斷器狀態(tài)機(jī)，狀態(tài)機(jī)的轉(zhuǎn)換關(guān)系如下圖所示：

熔斷器有三種狀態(tài)：

Closed狀態(tài)：也是初始狀態(tài)，我們需要一個(gè)調(diào)用失敗的計(jì)數(shù)器，如果調(diào)用失敗，則使失敗次數(shù)加1。如果最近失敗次數(shù)超過了在給定時(shí)間內(nèi)允許失敗的閾值，則切換到Open狀態(tài)，此時(shí)開啟一個(gè)超時(shí)時(shí)鐘，當(dāng)?shù)竭_(dá)超時(shí)時(shí)鐘時(shí)間后，則切換到HalfOpen狀態(tài)，該超時(shí)時(shí)間的設(shè)定是給了系統(tǒng)一次機(jī)會(huì)來修正導(dǎo)致調(diào)用失敗的錯(cuò)誤，以回到正常的工作狀態(tài)。在Closed狀態(tài)下，錯(cuò)誤計(jì)數(shù)是基于時(shí)間的。在特定的時(shí)間間隔內(nèi)會(huì)自動(dòng)重置，這能夠防止由于某次的偶然錯(cuò)誤導(dǎo)致熔斷器進(jìn)入Open狀態(tài)，也可以基于連續(xù)失敗的次數(shù)。Open狀態(tài)：在該狀態(tài)下，客戶端請求會(huì)立即返回錯(cuò)誤響應(yīng)，而不調(diào)用服務(wù)端。Half-Open狀態(tài)：允許客戶端一定數(shù)量的去調(diào)用服務(wù)端，如果這些請求對服務(wù)的調(diào)用成功，那么可以認(rèn)為之前導(dǎo)致調(diào)用失敗的錯(cuò)誤已經(jīng)修正，此時(shí)熔斷器切換到Closed狀態(tài)，同時(shí)將錯(cuò)誤計(jì)數(shù)器重置。如果這一定數(shù)量的請求有調(diào)用失敗的情況，則認(rèn)為導(dǎo)致之前調(diào)用失敗的的問題仍然存在，熔斷器切回到斷開狀態(tài)，然后重置計(jì)時(shí)器來給系統(tǒng)一定的時(shí)間來修正錯(cuò)誤。Half-Open狀態(tài)能夠有效防止正在恢復(fù)中的服務(wù)被突然而來的大量請求再次打掛。

下圖是Netflix的開源項(xiàng)目Hystrix中的熔斷器的實(shí)現(xiàn)邏輯：

從這個(gè)流程圖中，可以看到：

有請求來了，首先allowRequest()函數(shù)判斷是否在熔斷中，如果不是則放行，如果是的話，還要看有沒有達(dá)到一個(gè)熔斷時(shí)間片，如果熔斷時(shí)間片到了，也放行，否則直接返回錯(cuò)誤。每次調(diào)用都有兩個(gè)函數(shù)makeSuccess(duration)和makeFailure(duration)來統(tǒng)計(jì)一下在一定的duration內(nèi)有多少是成功還是失敗的。判斷是否熔斷的條件isOpen()，是計(jì)算failure/(success+failure)當(dāng)前的錯(cuò)誤率，如果高于一個(gè)閾值，那么熔斷器打開，否則關(guān)閉。Hystrix會(huì)在內(nèi)存中維護(hù)一個(gè)數(shù)據(jù)，其中記錄著每一個(gè)周期的請求結(jié)果的統(tǒng)計(jì)，超過時(shí)長長度的元素會(huì)被刪除掉。

熔斷器實(shí)現(xiàn)

了解了熔斷的原理后，我們來自己實(shí)現(xiàn)一套熔斷器。

熟悉go-zero的朋友都知道，在go-zero中熔斷沒有采用上面介紹的方式，而是參考了《GoogleSre》采用了一種自適應(yīng)的熔斷機(jī)制，這種自適應(yīng)的方式有什么好處呢？下文會(huì)基于這兩種機(jī)制做一個(gè)對比。

下面我們基于上面介紹的熔斷原理，實(shí)現(xiàn)一套自己的熔斷器。

代碼路徑：go-zero/core/breaker/hystrixbreaker.go

熔斷器默認(rèn)的狀態(tài)為Closed，當(dāng)熔斷器打開后默認(rèn)的冷卻時(shí)間是5秒鐘，當(dāng)熔斷器處于HalfOpen狀態(tài)時(shí)默認(rèn)的探測時(shí)間為200毫秒，默認(rèn)使用rateTripFunc方法來判斷是否觸發(fā)熔斷，規(guī)則是采樣大于等于200且錯(cuò)誤率大于50%，使用滑動(dòng)窗口來記錄請求總數(shù)和錯(cuò)誤數(shù)。

funcnewHystrixBreaker()*hystrixBreaker{

bucketDuration:=time.Duration(int64(window)/int64(buckets))

stat:=collection.NewRollingWindow(buckets,bucketDuration)

returnhystrixBreaker{

state:Closed,

coolingTimeout:defaultCoolingTimeout,

detectTimeout:defaultDetectTimeout,

tripFunc:rateTripFunc(defaultErrRate,defaultMinSample),

stat:stat,

now:time.Now,

funcrateTripFunc(ratefloat64,minSamplesint64)TripFunc{

returnfunc(rollingWindow*collection.RollingWindow)bool{

vartotal,errsint64

rollingWindow.Reduce(func(b*collection.Bucket){

total+=b.Count

errs+=int64(b.Sum)

errRate:=float64(errs)/float64(total)

returntotal=minSampleserrRaterate

每次請求都會(huì)調(diào)用doReq方法，在該方法中，首先通過accept()方法判斷是否拒絕本次請求，拒絕則直接返回熔斷錯(cuò)誤。否則執(zhí)行req()真正的發(fā)起服務(wù)端調(diào)用，成功和失敗分別調(diào)用b.markSuccess()和b.markFailure()

func(b*hystrixBreaker)doReq(reqfunc()error,fallbackfunc(error)error,acceptableAcceptable)error{

iferr:=b.accept();err!=nil{

iffallback!=nil{

returnfallback(err)

returnerr

deferfunc(){

ife:=recover();e!=nil{

b.markFailure()

panic(e)

err:=req()

ifacceptable(err){

b.markSuccess()

}else{

b.markFailure()

returnerr

在accept()方法中，首先獲取當(dāng)前熔斷器狀態(tài)，當(dāng)熔斷器處于Closed狀態(tài)直接返回，表示正常處理本次請求。

當(dāng)前狀態(tài)為Open的時(shí)候，判斷冷卻時(shí)間是否過期，如果沒有過期的話則直接返回熔斷錯(cuò)誤拒絕本次請求，如果過期的話則把熔斷器狀態(tài)更改為HalfOpen，冷卻時(shí)間的主要目的是給服務(wù)端一些時(shí)間進(jìn)行故障恢復(fù)，避免持續(xù)請求把服務(wù)端打掛。

當(dāng)前狀態(tài)為HalfOpen的時(shí)候，首先判斷探測時(shí)間間隔，避免探測過于頻繁，默認(rèn)使用200毫秒作為探測間隔。

func(b*hystrixBreaker)accept()error{

b.mux.Lock()

switchb.getState(){

caseOpen:

now:=b.now()

ifb.openTime.Add(b.coolingTimeout).After(now){

b.mux.Unlock()

returnErrServiceUnavailable

ifb.getState()==Open{

atomic.StoreInt32((*int32)(b.state),int32(HalfOpen))

atomic.StoreInt32(b.halfopenSuccess,0)

b.lastRetryTime=now

b.mux.Unlock()

}else{

b.mux.Unlock()

returnErrServiceUnavailable

caseHalfOpen:

now:=b.now()

ifb.lastRetryTime.Add(b.detectTimeout).After(now){

b.mux.Unlock()

returnErrServiceUnavailable

b.lastRetryTime=now

b.mux.Unlock()

caseClosed:

b.mux.Unlock()

returnnil

如果本次請求正常返回，則調(diào)用markSuccess()方法，如果當(dāng)前熔斷器處于HalfOpen狀態(tài)，則判斷當(dāng)前探測成功數(shù)量是否大于默認(rèn)的探測成功數(shù)量，如果大于則把熔斷器的狀態(tài)更新為Closed。

func(b*hystrixBreaker)markSuccess(){

b.mux.Lock()

switchb.getState(){

caseOpen:

b.mux.Unlock()

caseHalfOpen:

atomic.AddInt32(b.halfopenSuccess,1)

ifatomic.LoadInt32(b.halfopenSuccess)defaultHalfOpenSuccesss{

atomic.StoreInt32((*int32)(b.state),int32(Closed))

b.stat.Reduce(func(b*collection.Bucket){

b.Count=0

b.Sum=0

b.mux.Unlock()

caseClosed:

b.stat.Add(1)

b.mux.Unlock()

在markFailure()方法中，如果當(dāng)前狀態(tài)是Closed通過執(zhí)行tripFunc來判斷是否滿足熔斷條件，如果滿足則把熔斷器狀態(tài)更改為Open狀態(tài)。

func(b*hystrixBreaker)markFailure(){

b.mux.Lock()

b.stat.Add(0)

switchb.getState(){

caseOpen:

b.mux.Unlock()

caseHalfOpen:

b.openTime=b.now()

atomic.StoreInt32((*int32)(b.state),int32(Open))

b.mux.Unlock()

caseClosed:

ifb.tripFunc!=nilb.tripFunc(b.stat){

b.openTime=b.now()

atomic.StoreInt32((*int32)(b.state),int32(Open))

b.mux.Unlock()

熔斷器的實(shí)現(xiàn)邏輯總體比較簡單，閱讀代碼基本都能理解，這部分代碼實(shí)現(xiàn)的比較倉促，可能會(huì)有bug，如果大家發(fā)現(xiàn)bug可以隨時(shí)聯(lián)系我進(jìn)行修正。

hystrixBreaker和googlebreaker對比

接下來對比一下兩種熔斷器的熔斷效果。

這部分示例代碼在：go-zero/example下

分別定義了user-api和user-rpc服務(wù)，user-api作為客戶端對user-rpc進(jìn)行請求，user-rpc作為服務(wù)端響應(yīng)客戶端請求。

在user-rpc的示例方法中，有20%的幾率返回錯(cuò)誤。

func(l*UserInfoLogic)UserInfo(in*user.UserInfoRequest)(*user.UserInfoResponse,error){

ts:=time.Now().UnixMilli()

ifin.UserId==int64(1){

ifts%5==1{

returnnil,status.Error(codes.Internal,"internalerror")

returnuser.UserInfoResponse{

UserId:1,

Name:"jack",

},nil

returnuser.UserInfoResponse{},nil

在user-api的示例方法中，對user-rpc發(fā)起請求，然后使用prometheus指標(biāo)記錄正常請求的數(shù)量。

varmetricSuccessReqTotal=metric.NewCounterVec(metric.CounterVecOpts{

Namespace:"circuit_breaker",

Subsystem:"requests",

Name:"req_total",

Help:"testforcircuitbreaker",

Labels:[]string{"method"},

func(l*UserInfoLogic)UserInfo()(resp*types.UserInfoResponse,errerror){

for{

_,err:=l.svcCtx.UserRPC.UserInfo(l.ctx,user.UserInfoRequest{UserId:int64(1)})

iferr!=nilerr==breaker.ErrServiceUnavailable{

fmt.Println(err)

continue

metricSuccessReqTotal.Inc("UserInfo")

returntypes.UserInfoResponse{},nil

啟動(dòng)兩個(gè)服務(wù)，然后觀察在兩種熔斷策略下正常請求的數(shù)量。

googleBreaker熔斷器的正常請求率如下圖所示：

hystrixBreaker熔斷器的正常請求率如下圖所示：

從上面的實(shí)驗(yàn)結(jié)果可以看出，go-zero內(nèi)置的googleBreaker的正常請求數(shù)是高于hystrixBreaker的。這是因?yàn)閔ystrixBreaker維護(hù)了三種狀態(tài)，當(dāng)進(jìn)入Open狀態(tài)后為了避免繼續(xù)對服務(wù)端發(fā)起請求造成壓力，會(huì)使用一個(gè)冷卻時(shí)鐘，而在這段時(shí)間里是不會(huì)放過任何請求的，同時(shí)，從HalfOpen狀態(tài)變?yōu)镃losed狀態(tài)后，瞬間又會(huì)有大量的請求發(fā)往服務(wù)端，這時(shí)服務(wù)端很可能還沒恢復(fù)，從而導(dǎo)致熔斷器又變?yōu)镺pen狀態(tài)。

而googleBreaker采用的是一種自適應(yīng)的熔斷策略，也不需要多種狀態(tài)，也不會(huì)像hystrixBreaker那樣一刀切，而是會(huì)盡可能多的處理請求，這不也是我們期望的嘛，畢竟熔斷對客戶來說是有損的。下面我們來一起學(xué)習(xí)下go-zero內(nèi)置的熔斷器googleBreaker。

源碼解讀

googleBreaker的代碼路徑在：go-zero/core/breaker/googlebreaker.go

在doReq()方法中通過accept()方法判斷是否觸發(fā)熔斷，如果觸發(fā)熔斷則返回error，這里如果定義了回調(diào)函數(shù)的話可以執(zhí)行回調(diào)，比如做一些降級(jí)數(shù)據(jù)的處理等。如果請求正常則通過markSuccess()給總請求數(shù)和正常請求數(shù)都加1，如果請求失敗通過markFailure則只給總請求數(shù)加1。

func(b*googleBreaker)doReq(reqfunc()error,fallbackfunc(errerror)error,acceptableAcceptable)error{

iferr:=b.accept();err!=nil{

iffallback!=nil{

returnfallback(err)

returnerr

deferfunc(){

ife:=recover();e!=nil{

b.markFailure()

panic(e)

err:=req()

ifacceptable(err){

b.markSuccess()

}else{

b.markFailure()

returnerr

在accept()方法中通過計(jì)算判斷是否觸發(fā)熔斷。

在該算法中，需要記錄兩個(gè)請求數(shù)，分別是：

請求總量（requests）:調(diào)用方發(fā)起請求的數(shù)量總和正常處理的請求數(shù)量（accepts）:服務(wù)端正常處理的請求數(shù)量

在正常情況下，這兩個(gè)值是相等的，隨著被調(diào)用方服務(wù)出現(xiàn)異常開始拒絕請求，請求接受數(shù)量(accepts)的值開始逐漸小于請求數(shù)量(requests)，這個(gè)時(shí)候調(diào)用方可以繼續(xù)發(fā)送請求，直到requests=K*accepts，一旦超過這個(gè)限制，熔斷器就會(huì)打開，新的請求會(huì)在本地以一定的概率被拋棄直接返回錯(cuò)誤，概率的計(jì)算公式如下：

max(0,(requests-K*accepts)/(requests+1))

通過修改算法中的K(倍值)，可以調(diào)節(jié)熔斷器的敏感度，當(dāng)降低該倍值會(huì)使自適應(yīng)熔斷算法更敏感，當(dāng)增加該倍值會(huì)使得自適應(yīng)熔斷算法降低敏感度，舉例來說，假設(shè)將調(diào)用方的請求上限從requests=2acceptst調(diào)整為requests=1.1accepts那么就意味著調(diào)用方每十個(gè)請求之中就有一個(gè)請求會(huì)觸發(fā)熔斷。

func(b*googleBreaker)accept()error{

accepts,total:=b.history()

weightedAccepts:=b.k*float64(accepts)

///sre/sre-book/chapters/handling-overload/#e