关于微服务:微服务架构｜gozero-的自适应熔断器

原文链接： go-zero 的自适应熔断器

上篇文章咱们介绍了微服务的限流，详细分析了计数器限流和令牌桶限流算法，这篇文章来说说熔断。

熔断和限流还不太一样，限流是管制申请速率，只有还能接受，那么都会解决，但熔断不是。

在一条调用链上，如果发现某个服务异样，比方响应超时。那么调用者为了防止过多申请导致资源耗费过大，最终引发零碎雪崩，会间接返回谬误，而不是疯狂调用这个服务。

本篇文章会介绍支流熔断器的工作原理，并且会借助 go-zero 源码，剖析 googleBreaker 是如何通过滑动窗口来统计流量，并且最终执行熔断的。

工作原理

这部分次要介绍两种熔断器的工作原理，别离是 Netflix 开源的 Hystrix，其也是 Spring Cloud 默认的熔断组件，和 Google 的自适应的熔断器。

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

留神，Hystrix 官网曾经发表不再踊跃开发了，目前处在保护模式。

Hystrix 官网举荐代替的开源组件：Resilience4j，还有阿里开源的 Sentinel 也是不错的替代品。

hystrixBreaker

Hystrix 采纳了熔断器模式，相当于电路中的保险丝，零碎呈现紧急问题，立即禁止所有申请，已达到爱护零碎的作用。

零碎须要保护三种状态，别离是：

• 敞开： 默认状态，所有申请全副可能通过。当申请失败数量减少，失败率超过阈值时，会进入到断开状态。
• 断开： 此状态下，所有申请都会被拦挡。当通过一段超时工夫后，会进入到半断开状态。
• 半断开： 此状态下会容许一部分申请通过，并统计胜利数量，当申请胜利时，复原到敞开状态，否则持续断开。

通过状态的变更，能够无效避免零碎雪崩的问题。同时，在半断开状态下，又能够让零碎进行自我修复。

googleBreaker

googleBreaker 实现了一种自适应的熔断模式，来看一下算法的计算公式，客户端申请被回绝的概率。

参数很少，也比拟好了解：

1. requests：申请数量
2. accepts：后端接管的申请数量
3. K：敏感度，个别举荐 1.5-2 之间

通过剖析公式，咱们能够失去上面几个论断，也就是产生熔断的理论原理：

1. 失常状况下，requests 和 accepts 是相等的，回绝的概率就是 0，没有产生熔断
2. 当失常申请量，也就是 accepts 缩小时，概率会逐步减少，当概率大于 0 时，就会产生熔断。如果 accepts 等于 0 了，则齐全熔断。
3. 当服务复原后，requests 和 accepts 的数量会同时减少，但因为 K * accepts 增长的更快，所以概率又会很快变回到 0，相当于敞开了熔断。

总的来说，googleBreaker 的实现计划更加优雅，而且参数也少，不必保护那么多的状态。

go-zero 就是采纳了 googleBreaker 的计划，上面就来剖析代码，看看到底是怎么实现的。

接口设计

接口定义这部分我个人感觉还是挺不好了解的，看了好多遍才理清了它们之间的关系。

其实看代码和看书是一样的，书越看越薄，代码会越看越短。刚开始看感觉代码很长，随着看懂的中央越来越多，显著感觉代码变短了。所以遇到不懂的代码不要怕，重复看，总会看懂的。

首先来看一下 breaker 局部的 UML 图，有了这张图，很多中央看起来还是绝对清晰的，上面来详细分析。

这里用到了动态代理模式，也能够说是接口装璜器，接下来就看看到底是怎么定义的：

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

这个接口是最终实现熔断办法的接口，由 googleBreaker 构造体实现。

// 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),
    }
}

这个是实现了日志收集的构造体，首先它实现了 throttle 接口，而后它蕴含了一个字段 internalThrottle，相当于具体的熔断办法是代理给 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
    }))
}

所以当它执行相应办法时，都是间接调用 internalThrottle 接口的办法，而后再加上本人的逻辑。

这也就是代理所起到的作用，在不扭转原办法的根底上，扩大原办法的性能。

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

这就是它们之间的关系，如果感觉有点乱的话，就重复看，看的次数多了，就清晰了。

日志收集

上文介绍过了，loggedThrottle 是为了记录日志而设计的代理层，这部分内容来剖析一下是如何记录日志的。

// core/breaker/breaker.go
type errorWindow struct {
    // 记录日志的数组
    reasons [numHistoryReasons]string
    // 索引
    index   int
    // 数组元素数量，小于等于 numHistoryReasons
    count   int
    lock    sync.Mutex
}

func (ew *errorWindow) add(reason string) {ew.lock.Lock()
    // 记录谬误日志内容
    ew.reasons[ew.index] = fmt.Sprintf("%s %s", time.Now().Format(timeFormat), reason)
    // 对 numHistoryReasons 进行取余来失去数组索引
    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")
}

外围就是这里采纳了一个 环形数组，通过保护两个字段来实现，别离是 index 和 count。

count 示意数组中元素的个数，最大值是数组的长度；index 是索引，每次 +1，而后对数组长度取余失去新索引。

我之前有一次面试就让我设计一个环形数组，过后答的还不是很好，这次算是学会了。

滑动窗口

一般来说，想要判断是否须要触发熔断，那么首先要晓得一段时间的申请数量，一段时间内的数量统计能够应用 滑动窗口 来实现。

首先看一下滑动窗口的定义：

// core/collection/rollingwindow.go

type RollingWindow struct {
    lock          sync.RWMutex
    // 窗口大小
    size          int
    // 窗口数据容器
    win           *window
    // 工夫距离
    interval      time.Duration
    // 游标，用于定位以后应该写入哪个 bucket
    offset        int
    // 汇总数据时，是否疏忽以后正在写入桶的数据
    // 某些场景下因为以后正在写入的桶数据并没有通过残缺的窗口工夫距离
    // 可能导致以后桶的统计并不精确
    ignoreCurrent bool
    // 最初写入桶的工夫
    // 用于计算下一次写入数据距离最初一次写入数据的之间
    // 通过了多少个工夫距离
    lastTime      time.Duration // start time of the last bucket
}

再来看一下 window 的构造：

type Bucket struct {
    // 桶内值的和
    Sum   float64
    // 桶内 add 次数
    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 {
    // 桶，一个桶就是一个工夫距离
    buckets []*Bucket
    // 窗口大小，也就是桶的数量
    size    int
}

有了这两个构造之后，咱们就能够画出这个滑动窗口了，如图所示。

当初来看一下向窗口中增加数据，是怎么一个过程。

func (rw *RollingWindow) Add(v float64) {rw.lock.Lock()
    defer rw.lock.Unlock()
    // 获取以后写入下标
    rw.updateOffset()
    // 向 bucket 中写入数据
    rw.win.add(rw.offset, v)
}

func (rw *RollingWindow) span() int {
    // 计算间隔 lastTime 通过了多少个工夫距离，也就是多少个桶
    offset := int(timex.Since(rw.lastTime) / rw.interval)
    // 如果在窗口范畴内，返回理论值，否则返回窗口大小
    if 0 <= offset && offset < rw.size {return offset}

    return rw.size
}

func (rw *RollingWindow) updateOffset() {
    // 通过了多少个工夫距离，也就是多少个桶
    span := rw.span()
    // 还在同一单元工夫内不须要更新
    if span <= 0 {return}

    offset := rw.offset
    // reset expired buckets
    // 这里是革除过期桶的数据
    // 也是对数组大小进行取余的形式，相似上文介绍的环形数组
    for i := 0; i < span; i++ {rw.win.resetBucket((offset + i + 1) % rw.size)
    }

    // 更新游标
    rw.offset = (offset + span) % rw.size
    now := timex.Now()
    // align to interval time boundary
    // 这里应该是一个工夫的对齐，放弃在桶内指向地位是统一的
    rw.lastTime = now - (now-rw.lastTime)%rw.interval
}

// 向桶内增加数据
func (w *window) add(offset int, v float64) {
    // 依据 offset 对数组大小取余失去索引，而后增加数据
    w.buckets[offset%w.size].add(v)
}

// 重置桶数据
func (w *window) resetBucket(offset int) {w.buckets[offset%w.size].reset()}

我画了一张图，来模仿整个滑动过程：

次要经验 4 个步骤：

1. 计算以后工夫间隔上次增加工夫通过了多少个工夫距离，也就是多少个 bucket
2. 清理过期桶数据
3. 更新 offset，更新 offset 的过程理论就是模仿窗口滑动的过程
4. 增加数据

比方上图，刚开始 offset 指向了 bucket[1]，通过了两个 span 之后，bucket[2] 和 bucket[3] 会被清空，同时，新的 offset 会指向 bucket[3]，新增加的数据会写入到 bucket[3]。

再来看看数据统计，也就是窗口内的无效数据量是多少。

// 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}
    // 须要统计的 bucket 数量，窗口大小减去 span 数量
    if diff > 0 {
        // 获取统计的起始地位，span 是曾经被重置的 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++ {
        // 自定义统计函数
        fn(w.buckets[(start+i)%w.size])
    }
}

统计出窗口数据之后，就能够判断是否须要熔断了。

执行熔断

接下来就是执行熔断了，次要就是看看自适应熔断是如何实现的。

// core/breaker/googlebreaker.go

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

窗口的定义局部，整个窗口是 10s，而后分成 40 个 bucket，每个 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 {
    // 获取最近一段时间的统计数据
    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))
    // 如果小于等于 0 间接通过，不熔断
    if dropRatio <= 0 {return nil}

    // 随机产生 0.0-1.0 之间的随机数与下面计算出来的熔断概率相比拟
    // 如果随机数比熔断概率小则进行熔断
    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
}

以上就是自适应熔断的逻辑，通过概率的比拟来随机淘汰掉局部申请，而后随着服务复原，淘汰的申请会逐步变少，直至不淘汰。

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

    // 返回一个 promise 异步回调对象，可由开发者自行决定是否上报后果到熔断器
    return googlePromise{b: b,}, nil
}

// req - 熔断对象办法
// fallback - 自定义疾速失败函数，可对熔断产生的 err 进行包装后返回
// acceptable - 对本次未熔断时执行申请的后果进行自定义的断定，比方能够针对 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 则执行
        if fallback != nil {return fallback(err)
        }

        return err
    }

    defer func() {// 如果执行 req()过程产生了 panic，仍然断定本次执行失败上报至熔断器
        if e := recover(); e != nil {b.markFailure()
            panic(e)
        }
    }()

    err := req()
    // 上报后果
    if acceptable(err) {b.markSuccess()
    } else {b.markFailure()
    }

    return err
}

熔断器对外裸露两种类型的办法：

1、简略场景直接判断对象是否被熔断，执行申请后必须需手动上报执行后果至熔断器。

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

2、简单场景下反对自定义疾速失败，自定义断定申请是否胜利的熔断办法，主动上报执行后果至熔断器。

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

个人感觉，熔断这部分代码，相较于前几篇文章，了解起来是更艰难的。但其中的一些设计思维，和底层的实现原理也是十分值得学习的，心愿这篇文章可能对大家有帮忙。

以上就是本文的全部内容，如果感觉还不错的话欢送 点赞 ， 转发 和关注，感激反对。

参考文章：

• 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

举荐浏览：

• go-zero 是如何实现令牌桶限流的？
• go-zero 是如何实现计数器限流的？
• go-zero 是如何做路由治理的？