Go 項(xiàng)目中使用熔斷技術(shù)提高系統(tǒng)容錯(cuò)性。本文介紹了 Go 熔斷器和其使用。

[[436616]]

熔斷器像是一個(gè)保險(xiǎn)絲。當(dāng)我們依賴的服務(wù)出現(xiàn)問(wèn)題時(shí)，可以及時(shí)容錯(cuò)。一方面可以減少依賴服務(wù)對(duì)自身訪問(wèn)的依賴，防止出現(xiàn)雪崩效應(yīng)；另一方面降低請(qǐng)求頻率以方便上游盡快恢復(fù)服務(wù)。

熔斷器的應(yīng)用也非常廣泛。除了在我們應(yīng)用中，為了請(qǐng)求服務(wù)時(shí)使用熔斷器外，在 web 網(wǎng)關(guān)、微服務(wù)中，也有非常廣泛的應(yīng)用。本文將從源碼角度學(xué)習(xí) sony 開(kāi)源的一個(gè)熔斷器實(shí)現(xiàn) github/sony/gobreaker。（代碼注釋可以從github/lpflpf/gobreaker 查看)

熔斷器的模式

gobreaker 是基于《微軟云設(shè)計(jì)模式》一書(shū)中的熔斷器模式的 Golang 實(shí)現(xiàn)。有 sony 公司開(kāi)源，目前 star 數(shù)有 1.2K。使用人數(shù)較多。

下面是模式定義的一個(gè)狀態(tài)機(jī)：

熔斷器有三種狀態(tài)，四種狀態(tài)轉(zhuǎn)移的情況：

三種狀態(tài)：

•  熔斷器關(guān)閉狀態(tài)，服務(wù)正常訪問(wèn)
•  熔斷器開(kāi)啟狀態(tài)，服務(wù)異常
•  熔斷器半開(kāi)狀態(tài)，部分請(qǐng)求限流訪問(wèn)

四種狀態(tài)轉(zhuǎn)移：

•  在熔斷器關(guān)閉狀態(tài)下，當(dāng)失敗后并滿足一定條件后，將直接轉(zhuǎn)移為熔斷器開(kāi)啟狀態(tài)。
•  在熔斷器開(kāi)啟狀態(tài)下，如果過(guò)了規(guī)定的時(shí)間，將進(jìn)入半開(kāi)啟狀態(tài)，驗(yàn)證目前服務(wù)是否可用。
•  在熔斷器半開(kāi)啟狀態(tài)下，如果出現(xiàn)失敗，則再次進(jìn)入關(guān)閉狀態(tài)。
•  在熔斷器半開(kāi)啟后，所有請(qǐng)求（有限額）都是成功的，則熔斷器關(guān)閉。所有請(qǐng)求將正常訪問(wèn)。

gobreaker 的實(shí)現(xiàn)

gobreaker 是在上述狀態(tài)機(jī)的基礎(chǔ)上，實(shí)現(xiàn)的一個(gè)熔斷器。

熔斷器的定義 

type CircuitBreaker struct {   
  name          string    
  maxRequests   uint32  // 最大請(qǐng)求數(shù) （半開(kāi)啟狀態(tài)會(huì)限流）    
  interval      time.Duration   // 統(tǒng)計(jì)周期    
  timeout       time.Duration   // 進(jìn)入熔斷后的超時(shí)時(shí)間    
  readyToTrip   func(counts Counts) bool // 通過(guò) Counts 判斷是否開(kāi)啟熔斷。需要自定義    
  onStateChange func(name string, from State, to State) // 狀態(tài)修改時(shí)的鉤子函數(shù)   
  mutex      sync.Mutex // 互斥鎖，下面數(shù)據(jù)的更新都需要加鎖    
  state      State  // 記錄了當(dāng)前的狀態(tài)    
  generation uint64 // 標(biāo)記屬于哪個(gè)周期    
  counts     Counts // 計(jì)數(shù)器，統(tǒng)計(jì)了 成功、失敗、連續(xù)成功、連續(xù)失敗等，用于決策是否進(jìn)入熔斷    
  expiry     time.Time // 進(jìn)入下個(gè)周期的時(shí)間    
}   

其中，如下參數(shù)是我們可以自定義的：

•  MaxRequests：最大請(qǐng)求數(shù)。當(dāng)在最大請(qǐng)求數(shù)下，均請(qǐng)求正常的情況下，會(huì)關(guān)閉熔斷器
•  interval：一個(gè)正常的統(tǒng)計(jì)周期。如果為 0，那每次都會(huì)將計(jì)數(shù)清零
•  timeout: 進(jìn)入熔斷后，可以再次請(qǐng)求的時(shí)間
•  readyToTrip：判斷熔斷生效的鉤子函數(shù)
•  onStateChagne：狀態(tài)變更的鉤子函數(shù)

請(qǐng)求的執(zhí)行

熔斷器的執(zhí)行操作，主要包括三個(gè)階段；①請(qǐng)求之前的判定；②服務(wù)的請(qǐng)求執(zhí)行；③請(qǐng)求后的狀態(tài)和計(jì)數(shù)的更新 

// 熔斷器的調(diào)用    
func (cb *CircuitBreaker) Execute(req func() (interface{}, error)) (interface{}, error) {    
   // ①請(qǐng)求之前的判斷    
  generation, err := cb.beforeRequest()   
  if err != nil {    
    return nil, err    
  }    
  defer func() {   
    e := recover()   
    if e != nil {    
      // ③ panic 的捕獲    
      cb.afterRequest(generation, false)  
      panic(e)    
    }    
  }()    
  // ② 請(qǐng)求和執(zhí)行    
  result, err := req() 
   // ③ 更新計(jì)數(shù)   
  cb.afterRequest(generation, err == nil)    
  return result, err    
}   

請(qǐng)求之前的判定操作

請(qǐng)求之前，會(huì)判斷當(dāng)前熔斷器的狀態(tài)。如果熔斷器以開(kāi)啟，則不會(huì)繼續(xù)請(qǐng)求。如果熔斷器半開(kāi)，并且已達(dá)到最大請(qǐng)求閾值，也不會(huì)繼續(xù)請(qǐng)求。 

func (cb *CircuitBreaker) beforeRequest() (uint64, error) {    
  cb.mutex.Lock()    
  defer cb.mutex.Unlock()    
  now := time.Now()    
  state, generation := cb.currentState(now)    
    if state == StateOpen { // 熔斷器開(kāi)啟，直接返回    
    return generation, ErrOpenState    
  } else if state == StateHalfOpen && cb.counts.Requests >= cb.maxRequests { // 如果是半打開(kāi)的狀態(tài)，并且請(qǐng)求次數(shù)過(guò)多了，則直接返回    
    return generation, ErrTooManyRequests    
  }    
  cb.counts.onRequest()    
  return generation, nil   
}   

其中當(dāng)前狀態(tài)的計(jì)算，是依據(jù)當(dāng)前狀態(tài)來(lái)的。如果當(dāng)前狀態(tài)為已開(kāi)啟，則判斷是否已經(jīng)超時(shí)，超時(shí)就可以變更狀態(tài)到半開(kāi)；如果當(dāng)前狀態(tài)為關(guān)閉狀態(tài)，則通過(guò)周期判斷是否進(jìn)入下一個(gè)周期。 

func (cb *CircuitBreaker) currentState(now time.Time) (State, uint64) {    
  switch cb.state {    
  case StateClosed:    
    if !cb.expiry.IsZero() && cb.expiry.Before(now) { // 是否需要進(jìn)入下一個(gè)計(jì)數(shù)周期    
      cb.toNewGeneration(now)    
    }    
  case StateOpen:    
    if cb.expiry.Before(now) {    
      // 熔斷器由開(kāi)啟變更為半開(kāi)    
      cb.setState(StateHalfOpen, now)   
    }    
  }    
  return cb.state, cb.generation    
}   

周期長(zhǎng)度的設(shè)定，也是以據(jù)當(dāng)前狀態(tài)來(lái)的。如果當(dāng)前正常（熔斷器關(guān)閉），則設(shè)置為一個(gè) interval 的周期；如果當(dāng)前熔斷器是開(kāi)啟狀態(tài)，則設(shè)置為超時(shí)時(shí)間（超時(shí)后，才能變更為半開(kāi)狀態(tài)）。

請(qǐng)求之后的處理操作

每次請(qǐng)求之后，會(huì)通過(guò)請(qǐng)求結(jié)果是否成功，對(duì)熔斷器做計(jì)數(shù)。 

func (cb *CircuitBreaker) afterRequest(before uint64, success bool) {    
  cb.mutex.Lock()    
  defer cb.mutex.Unlock()     
  now := time.Now()      
  // 如果不在一個(gè)周期，就不再計(jì)數(shù)    
  state, generation := cb.currentState(now)   
  if generation != before {   
    return    
  }    
  if success {    
    cb.onSuccess(state, now)    
  } else {    
    cb.onFailure(state, now)    
  }    
}   

如果在半開(kāi)的狀態(tài)下：

•  如果請(qǐng)求成功，則會(huì)判斷當(dāng)前連續(xù)成功的請(qǐng)求數(shù) 大于等于 maxRequests， 則可以把狀態(tài)由半開(kāi)狀態(tài)轉(zhuǎn)移為關(guān)閉狀態(tài)
•  如果在半開(kāi)狀態(tài)下，請(qǐng)求失敗，則會(huì)直接將半開(kāi)狀態(tài)轉(zhuǎn)移為開(kāi)啟狀態(tài)

如果在關(guān)閉狀態(tài)下：

•  如果請(qǐng)求成功，則計(jì)數(shù)更新
•  如果請(qǐng)求失敗，則調(diào)用 readyToTrip 判斷是否需要將狀態(tài)關(guān)閉狀態(tài)轉(zhuǎn)移為開(kāi)啟狀態(tài)

總結(jié)

•  對(duì)于頻繁請(qǐng)求一些遠(yuǎn)程或者第三方的不可靠的服務(wù)，存在失敗的概率還是非常大的。使用熔斷器的好處就是可以是我們自身的服務(wù)不被這些不可靠的服務(wù)拖垮，造成雪崩。
•  由于熔斷器里面，不僅會(huì)維護(hù)不少的統(tǒng)計(jì)數(shù)據(jù)，還有互斥鎖做資源隔離，成本也會(huì)不少。
•  在半開(kāi)狀態(tài)下，可能出現(xiàn)請(qǐng)求過(guò)多的情況。這是由于半開(kāi)狀態(tài)下，連續(xù)請(qǐng)求成功的數(shù)量未達(dá)到最大請(qǐng)求值。所以，熔斷器對(duì)于請(qǐng)求時(shí)間過(guò)長(zhǎng)（但是比較頻繁）的服務(wù)可能會(huì)造成大量的 too many requests 錯(cuò)誤