原文鏈接： go-zero 的自適應(yīng)熔斷器

上篇文章我們介紹了微服務(wù)的限流，詳細(xì)分析了計(jì)數(shù)器限流和令牌桶限流算法，這篇文章來說說熔斷。

熔斷和限流還不太一樣，限流是控制請(qǐng)求速率，只要還能承受，那么都會(huì)處理，但熔斷不是。

在一條調(diào)用鏈上，如果發(fā)現(xiàn)某個(gè)服務(wù)異常，比如響應(yīng)超時(shí)。那么調(diào)用者為了避免過多請(qǐng)求導(dǎo)致資源消耗過大，最終引發(fā)系統(tǒng)雪崩，會(huì)直接返回錯(cuò)誤，而不是瘋狂調(diào)用這個(gè)服務(wù)。

本篇文章會(huì)介紹主流熔斷器的工作原理，并且會(huì)借助 go-zero 源碼，分析 googleBreaker 是如何通過滑動(dòng)窗口來統(tǒng)計(jì)流量，并且最終執(zhí)行熔斷的。

工作原理

這部分主要介紹兩種熔斷器的工作原理，分別是 Netflix 開源的 Hystrix，其也是 Spring Cloud 默認(rèn)的熔斷組件，和 Google 的自適應(yīng)的熔斷器。

Hystrix is no longer in active development, and is currently in maintenance mode.

注意，Hystrix 官方已經(jīng)宣布不再積極開發(fā)了，目前處在維護(hù)模式。

Hystrix 官方推薦替代的開源組件：Resilience4j，還有阿里開源的 Sentinel 也是不錯(cuò)的替代品。

hystrixBreaker

Hystrix 采用了熔斷器模式，相當(dāng)于電路中的保險(xiǎn)絲，系統(tǒng)出現(xiàn)緊急問題，立刻禁止所有請(qǐng)求，已達(dá)到保護(hù)系統(tǒng)的作用。

系統(tǒng)需要維護(hù)三種狀態(tài)，分別是：

• 關(guān)閉： 默認(rèn)狀態(tài)，所有請(qǐng)求全部能夠通過。當(dāng)請(qǐng)求失敗數(shù)量增加，失敗率超過閾值時(shí)，會(huì)進(jìn)入到斷開狀態(tài)。
• 斷開： 此狀態(tài)下，所有請(qǐng)求都會(huì)被攔截。當(dāng)經(jīng)過一段超時(shí)時(shí)間后，會(huì)進(jìn)入到半斷開狀態(tài)。
• 半斷開： 此狀態(tài)下會(huì)允許一部分請(qǐng)求通過，并統(tǒng)計(jì)成功數(shù)量，當(dāng)請(qǐng)求成功時(shí)，恢復(fù)到關(guān)閉狀態(tài)，否則繼續(xù)斷開。

通過狀態(tài)的變更，可以有效防止系統(tǒng)雪崩的問題。同時(shí)，在半斷開狀態(tài)下，又可以讓系統(tǒng)進(jìn)行自我修復(fù)。

googleBreaker

googleBreaker 實(shí)現(xiàn)了一種自適應(yīng)的熔斷模式，來看一下算法的計(jì)算公式，客戶端請(qǐng)求被拒絕的概率。

參數(shù)很少，也比較好理解：

1. requests：請(qǐng)求數(shù)量
2. accepts：后端接收的請(qǐng)求數(shù)量
3. K：敏感度，一般推薦 1.5-2 之間

通過分析公式，我們可以得到下面幾個(gè)結(jié)論，也就是產(chǎn)生熔斷的實(shí)際原理：

1. 正常情況下，requests 和 accepts 是相等的，拒絕的概率就是 0，沒有產(chǎn)生熔斷
2. 當(dāng)正常請(qǐng)求量，也就是 accepts 減少時(shí)，概率會(huì)逐漸增加，當(dāng)概率大于 0 時(shí)，就會(huì)產(chǎn)生熔斷。如果 accepts 等于 0 了，則完全熔斷。
3. 當(dāng)服務(wù)恢復(fù)后，requests 和 accepts 的數(shù)量會(huì)同時(shí)增加，但由于 K * accepts 增長的更快，所以概率又會(huì)很快變回到 0，相當(dāng)于關(guān)閉了熔斷。

總的來說，googleBreaker 的實(shí)現(xiàn)方案更加優(yōu)雅，而且參數(shù)也少，不用維護(hù)那么多的狀態(tài)。

go-zero 就是采用了 googleBreaker 的方案，下面就來分析代碼，看看到底是怎么實(shí)現(xiàn)的。

接口設(shè)計(jì)

接口定義這部分我個(gè)人感覺還是挺不好理解的，看了好多遍才理清了它們之間的關(guān)系。

其實(shí)看代碼和看書是一樣的，書越看越薄，代碼會(huì)越看越短。剛開始看感覺代碼很長，隨著看懂的地方越來越多，明顯感覺代碼變短了。所以遇到不懂的代碼不要怕，反復(fù)看，總會(huì)看懂的。

首先來看一下 breaker 部分的 UML 圖，有了這張圖，很多地方看起來還是相對(duì)清晰的，下面來詳細(xì)分析。

這里用到了靜態(tài)代理模式，也可以說是接口裝飾器，接下來就看看到底是怎么定義的：

// core/breaker/breaker.go
internalThrottle interface {
    allow() (internalPromise, error)
    doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error
}

// core/breaker/googlebreaker.go
type googleBreaker struct {
    k     float64
    stat  *collection.RollingWindow
    proba *mathx.Proba
}

這個(gè)接口是最終實(shí)現(xiàn)熔斷方法的接口，由 googleBreaker 結(jié)構(gòu)體實(shí)現(xiàn)。

// core/breaker/breaker.go
throttle interface {
    allow() (Promise, error)
    doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error
}

type loggedThrottle struct {
    name string
    internalThrottle
    errWin *errorWindow
}

func newLoggedThrottle(name string, t internalThrottle) loggedThrottle {
    return loggedThrottle{
        name:             name,
        internalThrottle: t,
        errWin:           new(errorWindow),
    }
}

這個(gè)是實(shí)現(xiàn)了日志收集的結(jié)構(gòu)體，首先它實(shí)現(xiàn)了 throttle 接口，然后它包含了一個(gè)字段 internalThrottle，相當(dāng)于具體的熔斷方法是代理給 internalThrottle 來做的。

// core/breaker/breaker.go
func (lt loggedThrottle) allow() (Promise, error) {
    promise, err := lt.internalThrottle.allow()
    return promiseWithReason{
        promise: promise,
        errWin:  lt.errWin,
    }, lt.logError(err)
}

func (lt loggedThrottle) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
    return lt.logError(lt.internalThrottle.doReq(req, fallback, func(err error) bool {
        accept := acceptable(err)
        if !accept && err != nil {
            lt.errWin.add(err.Error())
        }
        return accept
    }))
}

所以當(dāng)它執(zhí)行相應(yīng)方法時(shí)，都是直接調(diào)用 internalThrottle 接口的方法，然后再加上自己的邏輯。

這也就是代理所起到的作用，在不改變?cè)椒ǖ幕A(chǔ)上，擴(kuò)展原方法的功能。

// core/breaker/breaker.go
circuitBreaker struct {
    name string
    throttle
}

// NewBreaker returns a Breaker object.
// opts can be used to customize the Breaker.
func NewBreaker(opts ...Option) Breaker {
    var b circuitBreaker
    for _, opt := range opts {
        opt(&b)
    }
    if len(b.name) == 0 {
        b.name = stringx.Rand()
    }
    b.throttle = newLoggedThrottle(b.name, newGoogleBreaker())

    return &b
}

最終的熔斷器又將功能代理給了 throttle。

這就是它們之間的關(guān)系，如果感覺有點(diǎn)亂的話，就反復(fù)看，看的次數(shù)多了，就清晰了。

日志收集

上文介紹過了，loggedThrottle 是為了記錄日志而設(shè)計(jì)的代理層，這部分內(nèi)容來分析一下是如何記錄日志的。

// core/breaker/breaker.go
type errorWindow struct {
    // 記錄日志的數(shù)組
    reasons [numHistoryReasons]string
    // 索引
    index   int
    // 數(shù)組元素?cái)?shù)量，小于等于 numHistoryReasons
    count   int
    lock    sync.Mutex
}

func (ew *errorWindow) add(reason string) {
    ew.lock.Lock()
    // 記錄錯(cuò)誤日志內(nèi)容
    ew.reasons[ew.index] = fmt.Sprintf("%s %s", time.Now().Format(timeFormat), reason)
    // 對(duì) numHistoryReasons 進(jìn)行取余來得到數(shù)組索引
    ew.index = (ew.index + 1) % numHistoryReasons
    ew.count = mathx.MinInt(ew.count+1, numHistoryReasons)
    ew.lock.Unlock()
}

func (ew *errorWindow) String() string {
    var reasons []string

    ew.lock.Lock()
    // reverse order
    for i := ew.index - 1; i >= ew.index-ew.count; i-- {
        reasons = append(reasons, ew.reasons[(i+numHistoryReasons)%numHistoryReasons])
    }
    ew.lock.Unlock()

    return strings.Join(reasons, "\n")
}

核心就是這里采用了一個(gè)環(huán)形數(shù)組，通過維護(hù)兩個(gè)字段來實(shí)現(xiàn)，分別是 index 和 count。

count 表示數(shù)組中元素的個(gè)數(shù)，最大值是數(shù)組的長度；index 是索引，每次 +1，然后對(duì)數(shù)組長度取余得到新索引。

我之前有一次面試就讓我設(shè)計(jì)一個(gè)環(huán)形數(shù)組，當(dāng)時(shí)答的還不是很好，這次算是學(xué)會(huì)了。

滑動(dòng)窗口

一般來說，想要判斷是否需要觸發(fā)熔斷，那么首先要知道一段時(shí)間的請(qǐng)求數(shù)量，一段時(shí)間內(nèi)的數(shù)量統(tǒng)計(jì)可以使用滑動(dòng)窗口來實(shí)現(xiàn)。

首先看一下滑動(dòng)窗口的定義：

// core/collection/rollingwindow.go

type RollingWindow struct {
    lock          sync.RWMutex
    // 窗口大小
    size          int
    // 窗口數(shù)據(jù)容器
    win           *window
    // 時(shí)間間隔
    interval      time.Duration
    // 游標(biāo)，用于定位當(dāng)前應(yīng)該寫入哪個(gè) bucket
    offset        int
    // 匯總數(shù)據(jù)時(shí)，是否忽略當(dāng)前正在寫入桶的數(shù)據(jù)
    // 某些場(chǎng)景下因?yàn)楫?dāng)前正在寫入的桶數(shù)據(jù)并沒有經(jīng)過完整的窗口時(shí)間間隔
    // 可能導(dǎo)致當(dāng)前桶的統(tǒng)計(jì)并不準(zhǔn)確
    ignoreCurrent bool
    // 最后寫入桶的時(shí)間
    // 用于計(jì)算下一次寫入數(shù)據(jù)間隔最后一次寫入數(shù)據(jù)的之間
    // 經(jīng)過了多少個(gè)時(shí)間間隔
    lastTime      time.Duration // start time of the last bucket
}

再來看一下 window 的結(jié)構(gòu)：

type Bucket struct {
    // 桶內(nèi)值的和
    Sum   float64
    // 桶內(nèi) add 次數(shù)
    Count int64
}

func (b *Bucket) add(v float64) {
    b.Sum += v
    b.Count++
}

func (b *Bucket) reset() {
    b.Sum = 0
    b.Count = 0
}

type window struct {
    // 桶，一個(gè)桶就是一個(gè)時(shí)間間隔
    buckets []*Bucket
    // 窗口大小，也就是桶的數(shù)量
    size    int
}

有了這兩個(gè)結(jié)構(gòu)之后，我們就可以畫出這個(gè)滑動(dòng)窗口了，如圖所示。

現(xiàn)在來看一下向窗口中添加數(shù)據(jù)，是怎樣一個(gè)過程。

func (rw *RollingWindow) Add(v float64) {
    rw.lock.Lock()
    defer rw.lock.Unlock()
    // 獲取當(dāng)前寫入下標(biāo)
    rw.updateOffset()
    // 向 bucket 中寫入數(shù)據(jù)
    rw.win.add(rw.offset, v)
}

func (rw *RollingWindow) span() int {
    // 計(jì)算距離 lastTime 經(jīng)過了多少個(gè)時(shí)間間隔，也就是多少個(gè)桶
    offset := int(timex.Since(rw.lastTime) / rw.interval)
    // 如果在窗口范圍內(nèi)，返回實(shí)際值，否則返回窗口大小
    if 0 <= offset && offset < rw.size {
        return offset
    }

    return rw.size
}

func (rw *RollingWindow) updateOffset() {
    // 經(jīng)過了多少個(gè)時(shí)間間隔，也就是多少個(gè)桶
    span := rw.span()
    // 還在同一單元時(shí)間內(nèi)不需要更新
    if span <= 0 {
        return
    }

    offset := rw.offset
    // reset expired buckets
    // 這里是清除過期桶的數(shù)據(jù)
    // 也是對(duì)數(shù)組大小進(jìn)行取余的方式，類似上文介紹的環(huán)形數(shù)組
    for i := 0; i < span; i++ {
        rw.win.resetBucket((offset + i + 1) % rw.size)
    }

    // 更新游標(biāo)
    rw.offset = (offset + span) % rw.size
    now := timex.Now()
    // align to interval time boundary
    // 這里應(yīng)該是一個(gè)時(shí)間的對(duì)齊，保持在桶內(nèi)指向位置是一致的
    rw.lastTime = now - (now-rw.lastTime)%rw.interval
}

// 向桶內(nèi)添加數(shù)據(jù)
func (w *window) add(offset int, v float64) {
    // 根據(jù) offset 對(duì)數(shù)組大小取余得到索引，然后添加數(shù)據(jù)
    w.buckets[offset%w.size].add(v)
}

// 重置桶數(shù)據(jù)
func (w *window) resetBucket(offset int) {
    w.buckets[offset%w.size].reset()
}

我畫了一張圖，來模擬整個(gè)滑動(dòng)過程：

主要經(jīng)歷 4 個(gè)步驟：

1. 計(jì)算當(dāng)前時(shí)間距離上次添加時(shí)間經(jīng)過了多少個(gè)時(shí)間間隔，也就是多少個(gè) bucket
2. 清理過期桶數(shù)據(jù)
3. 更新 offset，更新 offset 的過程實(shí)際就是模擬窗口滑動(dòng)的過程
4. 添加數(shù)據(jù)

比如上圖，剛開始 offset 指向了 bucket[1]，經(jīng)過了兩個(gè) span 之后，bucket[2] 和 bucket[3] 會(huì)被清空，同時(shí)，新的 offset 會(huì)指向 bucket[3]，新添加的數(shù)據(jù)會(huì)寫入到 bucket[3]。

再來看看數(shù)據(jù)統(tǒng)計(jì)，也就是窗口內(nèi)的有效數(shù)據(jù)量是多少。

// Reduce runs fn on all buckets, ignore current bucket if ignoreCurrent was set.
func (rw *RollingWindow) Reduce(fn func(b *Bucket)) {
    rw.lock.RLock()
    defer rw.lock.RUnlock()

    var diff int
    span := rw.span()
    // ignore current bucket, because of partial data
    if span == 0 && rw.ignoreCurrent {
        diff = rw.size - 1
    } else {
        diff = rw.size - span
    }
    // 需要統(tǒng)計(jì)的 bucket 數(shù)量，窗口大小減去 span 數(shù)量
    if diff > 0 {
        // 獲取統(tǒng)計(jì)的起始位置，span 是已經(jīng)被重置的 bucket
        offset := (rw.offset + span + 1) % rw.size
        rw.win.reduce(offset, diff, fn)
    }
}

func (w *window) reduce(start, count int, fn func(b *Bucket)) {
    for i := 0; i < count; i++ {
        // 自定義統(tǒng)計(jì)函數(shù)
        fn(w.buckets[(start+i)%w.size])
    }
}

統(tǒng)計(jì)出窗口數(shù)據(jù)之后，就可以判斷是否需要熔斷了。

執(zhí)行熔斷

接下來就是執(zhí)行熔斷了，主要就是看看自適應(yīng)熔斷是如何實(shí)現(xiàn)的。

// core/breaker/googlebreaker.go

const (
    // 250ms for bucket duration
    window     = time.Second * 10
    buckets    = 40
    k          = 1.5
    protection = 5
)

窗口的定義部分，整個(gè)窗口是 10s，然后分成 40 個(gè) bucket，每個(gè) bucket 就是 250ms。

// googleBreaker is a netflixBreaker pattern from google.
// see Client-Side Throttling section in https://landing.google.com/sre/sre-book/chapters/handling-overload/
type googleBreaker struct {
    k     float64
    stat  *collection.RollingWindow
    proba *mathx.Proba
}

func (b *googleBreaker) accept() error {
    // 獲取最近一段時(shí)間的統(tǒng)計(jì)數(shù)據(jù)
    accepts, total := b.history()
    // 根據(jù)上文提到的算法來計(jì)算一個(gè)概率
    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))
    // 如果小于等于 0 直接通過，不熔斷
    if dropRatio <= 0 {
        return nil
    }

    // 隨機(jī)產(chǎn)生 0.0-1.0 之間的隨機(jī)數(shù)與上面計(jì)算出來的熔斷概率相比較
    // 如果隨機(jī)數(shù)比熔斷概率小則進(jìn)行熔斷
    if b.proba.TrueOnProba(dropRatio) {
        return ErrServiceUnavailable
    }

    return nil
}

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

    return
}

以上就是自適應(yīng)熔斷的邏輯，通過概率的比較來隨機(jī)淘汰掉部分請(qǐng)求，然后隨著服務(wù)恢復(fù)，淘汰的請(qǐng)求會(huì)逐漸變少，直至不淘汰。

func (b *googleBreaker) allow() (internalPromise, error) {
    if err := b.accept(); err != nil {
        return nil, err
    }

    // 返回一個(gè) promise 異步回調(diào)對(duì)象，可由開發(fā)者自行決定是否上報(bào)結(jié)果到熔斷器
    return googlePromise{
        b: b,
    }, nil
}

// req - 熔斷對(duì)象方法
// fallback - 自定義快速失敗函數(shù)，可對(duì)熔斷產(chǎn)生的err進(jìn)行包裝后返回
// acceptable - 對(duì)本次未熔斷時(shí)執(zhí)行請(qǐng)求的結(jié)果進(jìn)行自定義的判定，比如可以針對(duì)http.code,rpc.code,body.code
func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
    if err := b.accept(); err != nil {
        // 熔斷中，如果有自定義的fallback則執(zhí)行
        if fallback != nil {
            return fallback(err)
        }

        return err
    }

    defer func() {
        // 如果執(zhí)行req()過程發(fā)生了panic，依然判定本次執(zhí)行失敗上報(bào)至熔斷器
        if e := recover(); e != nil {
            b.markFailure()
            panic(e)
        }
    }()

    err := req()
    // 上報(bào)結(jié)果
    if acceptable(err) {
        b.markSuccess()
    } else {
        b.markFailure()
    }

    return err
}

熔斷器對(duì)外暴露兩種類型的方法：

1、簡單場(chǎng)景直接判斷對(duì)象是否被熔斷，執(zhí)行請(qǐng)求后必須需手動(dòng)上報(bào)執(zhí)行結(jié)果至熔斷器。

func (b *googleBreaker) allow() (internalPromise, error)

2、復(fù)雜場(chǎng)景下支持自定義快速失敗，自定義判定請(qǐng)求是否成功的熔斷方法，自動(dòng)上報(bào)執(zhí)行結(jié)果至熔斷器。

func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error

個(gè)人感覺，熔斷這部分代碼，相較于前幾篇文章，理解起來是更困難的。但其中的一些設(shè)計(jì)思想，和底層的實(shí)現(xiàn)原理也是非常值得學(xué)習(xí)的，希望這篇文章能夠?qū)Υ蠹矣袔椭?/p>

以上就是本文的全部內(nèi)容，如果覺得還不錯(cuò)的話歡迎點(diǎn)贊，轉(zhuǎn)發(fā)和關(guān)注，感謝支持。

參考文章：

• https://juejin.cn/post/7030997067560386590
• https://go-zero.dev/docs/tutorials/service/governance/breaker
• https://sre.google/sre-book/handling-overload/
• https://martinfowler.com/bliki/CircuitBreaker.html

推薦閱讀：文章來源地址http://www.zghlxwxcb.cn/news/detail-689732.html

• go-zero 是如何實(shí)現(xiàn)令牌桶限流的？
• go-zero 是如何實(shí)現(xiàn)計(jì)數(shù)器限流的？
• go-zero 是如何做路由管理的？

到了這里，關(guān)于微服務(wù)架構(gòu)｜go-zero 的自適應(yīng)熔斷器的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！