日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

學無先后，達者為師

網站首頁 編程語言 正文

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

作者：萬俊峰Kevin ? 更新時間： 2022-10-27 編程語言

正文

熔斷機制（Circuit Breaker）指的是在股票市場的交易時間中，當價格的波動幅度達到某一個限定的目標（熔斷點）時，對其暫停交易一段時間的機制。此機制如同保險絲在電流過大時候熔斷，故而得名。熔斷機制推出的目的是為了防范系統性風險，給市場更多的冷靜時間，避免恐慌情緒蔓延導致整個市場波動，從而防止大規模股價下跌現象的發生。

同樣的，在高并發的分布式系統設計中，也應該有熔斷的機制。熔斷一般是在客戶端（調用端）進行配置，當客戶端向服務端發起請求的時候，服務端的錯誤不斷地增多，這時候就可能會觸發熔斷，觸發熔斷后客戶端的請求不再發往服務端，而是在客戶端直接拒絕請求，從而可以保護服務端不會過載。這里說的服務端可能是rpc服務，http服務，也可能是mysql，redis等。注意熔斷是一種有損的機制，當熔斷后可能需要一些降級的策略進行配合。

熔斷原理

現代微服務架構基本都是分布式的，整個分布式系統是由非常多的微服務組成。不同服務之間相互調用，組成復雜的調用鏈路。在復雜的調用鏈路中的某一個服務如果不穩定，就可能會層層級聯，最終可能導致整個鏈路全部掛掉。因此我們需要對不穩定的服務依賴進行熔斷降級，暫時切斷不穩定的服務調用，避免局部不穩定因素導致整個分布式系統的雪崩。

說白了，我覺得熔斷就像是那些容易異常服務的一種代理，這個代理能夠記錄最近調用發生錯誤的次數，然后決定是繼續操作，還是立即返回錯誤。

熔斷器內部維護了一個熔斷器狀態機，狀態機的轉換關系如下圖所示：

熔斷器有三種狀態：

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

下圖是Netflix的開源項目Hystrix中的熔斷器的實現邏輯：

從這個流程圖中，可以看到：

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

熔斷器實現

了解了熔斷的原理后，我們來自己實現一套熔斷器。

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

下面我們基于上面介紹的熔斷原理，實現一套自己的熔斷器。

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

熔斷器默認的狀態為Closed，當熔斷器打開后默認的冷卻時間是5秒鐘，當熔斷器處于HalfOpen狀態時默認的探測時間為200毫秒，默認使用rateTripFunc方法來判斷是否觸發熔斷，規則是采樣大于等于200且錯誤率大于50%，使用滑動窗口來記錄請求總數和錯誤數。

func newHystrixBreaker() *hystrixBreaker {
  bucketDuration := time.Duration(int64(window) / int64(buckets))
  stat := collection.NewRollingWindow(buckets, bucketDuration)
  return &hystrixBreaker{
    state:          Closed,
    coolingTimeout: defaultCoolingTimeout,
    detectTimeout:  defaultDetectTimeout,
    tripFunc:       rateTripFunc(defaultErrRate, defaultMinSample),
    stat:           stat,
    now:            time.Now,
  }
}

func rateTripFunc(rate float64, minSamples int64) TripFunc {
  return func(rollingWindow *collection.RollingWindow) bool {
    var total, errs int64
    rollingWindow.Reduce(func(b *collection.Bucket) {
      total += b.Count
      errs += int64(b.Sum)
    })
    errRate := float64(errs) / float64(total)
    return total >= minSamples && errRate > rate
  }
}

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

func (b *hystrixBreaker) doReq(req func() error, fallback func(error) error, acceptable Acceptable) error {
  if err := b.accept(); err != nil {
    if fallback != nil {
      return fallback(err)
    }
    return err
  }
  defer func() {
    if e := recover(); e != nil {
      b.markFailure()
      panic(e)
    }
  }()
  err := req()
  if acceptable(err) {
    b.markSuccess()
  } else {
    b.markFailure()
  }
  return err
}

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

當前狀態為Open的時候，判斷冷卻時間是否過期，如果沒有過期的話則直接返回熔斷錯誤拒絕本次請求，如果過期的話則把熔斷器狀態更改為HalfOpen，冷卻時間的主要目的是給服務端一些時間進行故障恢復，避免持續請求把服務端打掛。

當前狀態為HalfOpen的時候，首先判斷探測時間間隔，避免探測過于頻繁，默認使用200毫秒作為探測間隔。

func (b *hystrixBreaker) accept() error {
  b.mux.Lock()
  switch b.getState() {
  case Open:
    now := b.now()
    if b.openTime.Add(b.coolingTimeout).After(now) {
      b.mux.Unlock()
      return ErrServiceUnavailable
    }
    if b.getState() == Open {
      atomic.StoreInt32((*int32)(&b.state), int32(HalfOpen))
      atomic.StoreInt32(&b.halfopenSuccess, 0)
      b.lastRetryTime = now
      b.mux.Unlock()
    } else {
      b.mux.Unlock()
      return ErrServiceUnavailable
    }
  case HalfOpen:
    now := b.now()
    if b.lastRetryTime.Add(b.detectTimeout).After(now) {
      b.mux.Unlock()
      return ErrServiceUnavailable
    }
    b.lastRetryTime = now
    b.mux.Unlock()
  case Closed:
    b.mux.Unlock()
  }
  return nil
}

如果本次請求正常返回，則調用markSuccess()方法，如果當前熔斷器處于HalfOpen狀態，則判斷當前探測成功數量是否大于默認的探測成功數量，如果大于則把熔斷器的狀態更新為Closed。

func (b *hystrixBreaker) markSuccess() {
  b.mux.Lock()
  switch b.getState() {
  case Open:
    b.mux.Unlock()
  case HalfOpen:
    atomic.AddInt32(&b.halfopenSuccess, 1)
    if atomic.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()
  case Closed:
    b.stat.Add(1)
    b.mux.Unlock()
  }
}

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

func (b *hystrixBreaker) markFailure() {
  b.mux.Lock()
  b.stat.Add(0)
  switch b.getState() {
  case Open:
    b.mux.Unlock()
  case HalfOpen:
    b.openTime = b.now()
    atomic.StoreInt32((*int32)(&b.state), int32(Open))
    b.mux.Unlock()
  case Closed:
    if b.tripFunc != nil && b.tripFunc(b.stat) {
      b.openTime = b.now()
      atomic.StoreInt32((*int32)(&b.state), int32(Open))
    }
    b.mux.Unlock()
  }
}

熔斷器的實現邏輯總體比較簡單，閱讀代碼基本都能理解，這部分代碼實現的比較倉促，可能會有bug，如果大家發現bug可以隨時聯系我進行修正。

hystrixBreaker和googlebreaker對比

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

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

分別定義了user-api和user-rpc服務，user-api作為客戶端對user-rpc進行請求，user-rpc作為服務端響應客戶端請求。

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

func (l *UserInfoLogic) UserInfo(in *user.UserInfoRequest) (*user.UserInfoResponse, error) {
  ts := time.Now().UnixMilli()
  if in.UserId == int64(1) {
    if ts%5 == 1 {
      return nil, status.Error(codes.Internal, "internal error")
    }
    return &user.UserInfoResponse{
      UserId: 1,
      Name:   "jack",
    }, nil
  }
  return &user.UserInfoResponse{}, nil
}

在user-api的示例方法中，對user-rpc發起請求，然后使用prometheus指標記錄正常請求的數量。

var metricSuccessReqTotal = metric.NewCounterVec(&metric.CounterVecOpts{
  Namespace: "circuit_breaker",
  Subsystem: "requests",
  Name:      "req_total",
  Help:      "test for circuit breaker",
  Labels:    []string{"method"},
})
func (l *UserInfoLogic) UserInfo() (resp *types.UserInfoResponse, err error) {
  for {
    _, err := l.svcCtx.UserRPC.UserInfo(l.ctx, &user.UserInfoRequest{UserId: int64(1)})
    if err != nil && err == breaker.ErrServiceUnavailable {
      fmt.Println(err)
      continue
    }
    metricSuccessReqTotal.Inc("UserInfo")
  }
  return &types.UserInfoResponse{}, nil
}

啟動兩個服務，然后觀察在兩種熔斷策略下正常請求的數量。

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

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

從上面的實驗結果可以看出，go-zero內置的googleBreaker的正常請求數是高于hystrixBreaker的。這是因為hystrixBreaker維護了三種狀態，當進入Open狀態后為了避免繼續對服務端發起請求造成壓力，會使用一個冷卻時鐘，而在這段時間里是不會放過任何請求的，同時，從HalfOpen狀態變為Closed狀態后，瞬間又會有大量的請求發往服務端，這時服務端很可能還沒恢復，從而導致熔斷器又變為Open狀態。

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

源碼解讀

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

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

func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
  if err := b.accept(); err != nil {
    if fallback != nil {
      return fallback(err)
    }
    return err
  }
  defer func() {
    if e := recover(); e != nil {
      b.markFailure()
      panic(e)
    }
  }()
  err := req()
  if acceptable(err) {
    b.markSuccess()
  } else {
    b.markFailure()
  }
  return err
}

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

在該算法中，需要記錄兩個請求數，分別是：

  • 請求總量（requests）: 調用方發起請求的數量總和
  • 正常處理的請求數量（accepts）: 服務端正常處理的請求數量

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

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

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

func (b *googleBreaker) accept() error {
  accepts, total := b.history()
  weightedAccepts := b.k * float64(accepts)
  // https://landing.google.com/sre/sre-book/chapters/handling-overload/#eq2101
  dropRatio := math.Max(0, (float64(total-protection)-weightedAccepts)/float64(total+1))
  if dropRatio <= 0 {
    return nil
  }
  if b.proba.TrueOnProba(dropRatio) {
    return ErrServiceUnavailable
  }
  return nil
}

history從滑動窗口中統計當前的總請求數和正常處理的請求數。

func (b *googleBreaker) history() (accepts, total int64) {
  b.stat.Reduce(func(b *collection.Bucket) {
    accepts += int64(b.Sum)
    total += b.Count
  })
  return
}

結束語

本篇文章介紹了服務治理中的一種客戶端節流機制 - 熔斷。在hystrix熔斷策略中需要實現三個狀態，分別是Open、HalfOpen和Closed。不同狀態的切換時機在上文中也有詳細描述，大家可以反復閱讀理解，最好是能自己動手實現一下。對于go-zero內置的熔斷器是沒有狀態的，如果非要說它的狀態的話，那么也只有打開和關閉兩種情況，它是根據當前請求的成功率自適應的丟棄請求，是一種更彈性的熔斷策略，丟棄請求概率隨著正常處理的請求數不斷變化，正常處理的請求越多丟棄請求的概率就越低，反之丟棄請求的概率就越高。

雖然熔斷的原理都一樣，但實現機制不同導致的效果可能也不同，在實際生產中可以根據實際情況選擇符合業務場景的熔斷策略。

希望本篇文章對你有所幫助。

本篇文章代碼：github.com/zhoushuguan…

參考

martinfowler.com/bliki/Circu…

github.com/Netflix/Hys…

項目地址

github.com/zeromicro/g…

原文鏈接：https://juejin.cn/post/7137836515803004964

欄目分類
最近更新