一文讲透自适应熔断的原理和实现

 // 自定义判定执行结果 Acceptable func(err error) bool  // 手动回调 Promise interface {  // Accept tells the Breaker that the call is successful.  // 请求成功  Accept()  // Reject tells the Breaker that the call is failed.  // 请求失败  Reject(reason string) } 
 Breaker interface {  // 熔断器名称  Name() string
  // 熔断方法，执行请求时必须手动上报执行结果  // 适用于简单无需自定义快速失败，无需自定义判定请求结果的场景  // 相当于手动挡。。。  Allow() (Promise, error)
  // 熔断方法，自动上报执行结果  // 自动挡。。。  Do(req func() error) error
  // 熔断方法  // acceptable - 支持自定义判定执行结果  DoWithAcceptable(req func() error, acceptable Acceptable) error
  // 熔断方法  // fallback - 支持自定义快速失败  DoWithFallback(req func() error, fallback func(err error) error) error
  // 熔断方法  // fallback - 支持自定义快速失败  // acceptable - 支持自定义判定执行结果  DoWithFallbackAcceptable(req func() error, fallback func(err error) error, acceptable Acceptable) error }

// 熔断器结构体circuitBreaker struct { name string // 实际上 circuitBreaker熔断功能都代理给 throttle来实现 throttle}// 熔断器接口throttle interface { // 熔断方法 allow() (Promise, error) // 熔断方法 // DoXXX()方法最终都会该方法 doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error} func (cb *circuitBreaker) Allow() (Promise, error) {  return cb.throttle.allow()}    func (cb *circuitBreaker) Do(req func() error) error {  return cb.throttle.doReq(req, nil, defaultAcceptable)}    func (cb *circuitBreaker) DoWithAcceptable(req func() error, acceptable Acceptable) error {  return cb.throttle.doReq(req, nil, acceptable)}    func (cb *circuitBreaker) DoWithFallback(req func() error, fallback func(err error) error) error {  return cb.throttle.doReq(req, fallback, defaultAcceptable)}    func (cb *circuitBreaker) DoWithFallbackAcceptable(req func() error, fallback func(err error) error,  acceptable Acceptable) error {    return cb.throttle.doReq(req, fallback, acceptable)}  

// 带日志功能的熔断器type loggedThrottle struct { // 名称 name string // 代理对象 internalThrottle // 滚动窗口,滚动收集数据,相当于环形数组 errWin *errorWindow}
// 熔断方法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 {   lt.errWin.add(err.Error())  }  return accept }))}
func (lt loggedThrottle) logError(err error) error { if err == ErrServiceUnavailable {  // if circuit open, not possible to have empty error window  stat.Report(fmt.Sprintf(   "proc(%s/%d), callee: %s, breaker is open and requests dropped\nlast errors:\n%s",   proc.ProcessName(), proc.Pid(), lt.name, lt.errWin)) }
 return err}

// 滚动窗口type errorWindow struct { reasons [numHistoryReasons]string index   int count   int lock    sync.Mutex}
// 添加数据func (ew *errorWindow) add(reason string) { ew.lock.Lock() // 添加错误日志 ew.reasons[ew.index] = fmt.Sprintf("%s %s", timex.Time().Format(timeFormat), reason) // 更新index,为下一次写入数据做准备 // 这里用的取模实现了滚动功能 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")}

type internalThrottle interface {  allow() (internalPromise, error)  doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error}

type googleBreaker struct { // 敏感度，go-zero中默认值为1.5 k float64 // 滑动窗口，用于记录最近一段时间内的请求总数，成功总数 stat *collection.RollingWindow // 概率生成器 // 随机产生0.0-1.0之间的双精度浮点数 proba *mathx.Proba}

RollingWindow struct {  // 互斥锁  lock sync.RWMutex  // 滑动窗口数量  size int  // 窗口，数据容器  win *window  // 滑动窗口单元时间间隔  interval time.Duration  // 游标，用于定位当前应该写入哪个bucket  offset int  // 汇总数据时，是否忽略当前正在写入桶的数据  // 某些场景下因为当前正在写入的桶数据并没有经过完整的窗口时间间隔  // 可能导致当前桶的统计并不准确  ignoreCurrent bool  // 最后写入桶的时间  // 用于计算下一次写入数据间隔最后一次写入数据的之间  // 经过了多少个时间间隔  lastTime      time.Duration }

// 时间窗口type window struct { // 桶 // 一个桶标识一个时间间隔 buckets []*Bucket // 窗口大小 size int}
// 添加数据// offset - 游标，定位写入bucket位置// v - 行为数据func (w *window) add(offset int, v float64) { w.buckets[offset%w.size].add(v)}
// 汇总数据// fn - 自定义的bucket统计函数func (w *window) reduce(start, count int, fn func(b *Bucket)) { for i := 0; i < count; i++ {  fn(w.buckets[(start+i)%w.size]) }}
// 清理特定bucketfunc (w *window) resetBucket(offset int) { w.buckets[offset%w.size].reset()}
// 桶type Bucket struct { // 当前桶内值之和 Sum float64 // 当前桶的add总次数 Count int64}
// 向桶添加数据func (b *Bucket) add(v float64) { // 求和 b.Sum += v // 次数+1 b.Count++}
// 桶数据清零func (b *Bucket) reset() { b.Sum = 0 b.Count = 0}

// 添加数据func (rw *RollingWindow) Add(v float64) { rw.lock.Lock() defer rw.lock.Unlock() // 获取当前写入的下标 rw.updateOffset() // 添加数据 rw.win.add(rw.offset, v)}
// 计算当前距离最后写入数据经过多少个单元时间间隔// 实际上指的就是经过多少个桶func (rw *RollingWindow) span() int { offset := int(timex.Since(rw.lastTime) / rw.interval) if 0 <= offset && offset < rw.size {  return offset } // 大于时间窗口时 返回窗口大小即可 return rw.size}
// 更新当前时间的offset// 实现窗口滑动func (rw *RollingWindow) updateOffset() { // 经过span个桶的时间 span := rw.span() // 还在同一单元时间内不需要更新 if span <= 0 {  return } offset := rw.offset // 既然经过了span个桶的时间没有写入数据 // 那么这些桶内的数据就不应该继续保留了，属于过期数据清空即可 // 可以看到这里全部用的 % 取余操作，可以实现按照下标周期性写入 // 如果超出下标了那就从头开始写，确保新数据一定能够正常写入 // 类似循环数组的效果 for i := 0; i < span; i++ {  rw.win.resetBucket((offset + i + 1) % rw.size) } // 更新offset rw.offset = (offset + span) % rw.size now := timex.Now() // 更新操作时间 // 这里很有意思 rw.lastTime = now - (now-rw.lastTime)%rw.interval}

// 归纳汇总数据func (rw *RollingWindow) Reduce(fn func(b *Bucket)) { rw.lock.RLock() defer rw.lock.RUnlock()
 var diff int span := rw.span() // 当前时间截止前，未过期桶的数量 if span == 0 && rw.ignoreCurrent {  diff = rw.size - 1 } else {  diff = rw.size - span } if diff > 0 {  // rw.offset - rw.offset+span之间的桶数据是过期的不应该计入统计  offset := (rw.offset + span + 1) % rw.size  // 汇总数据  rw.win.reduce(offset, diff, fn) }}

// 按照最近一段时间的请求数据计算是否熔断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}

type Acceptable func(err error) bool

// 熔断方法// 返回一个promise异步回调对象，可由开发者自行决定是否上报结果到熔断器func (b *googleBreaker) allow() (internalPromise, error) { if err := b.accept(); err != nil {  return nil, err }
 return googlePromise{  b: b, }, nil}
// 熔断方法// req - 熔断对象方法// fallback - 自定义快速失败函数，可对熔断产生的err进行包装后返回// acceptable - 对本次未熔断时执行请求的结果进行自定义的判定，比如可以针对http.code,rpc.code,body.codefunc (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 } // 如果执行req()过程发生了panic，依然判定本次执行失败上报至熔断器 defer func() {  if e := recover(); e != nil {   b.markFailure()   panic(e)  } }() // 执行请求 err := req() // 判定请求成功 if acceptable(err) {  b.markSuccess() } else {  b.markFailure() }
 return err}
// 上报成功func (b *googleBreaker) markSuccess() { b.stat.Add(1)}
// 上报失败func (b *googleBreaker) markFailure() { b.stat.Add(0)}
// 统计数据func (b *googleBreaker) history() (accepts, total int64) { b.stat.Reduce(func(b *collection.Bucket) {  accepts += int64(b.Sum)  total += b.Count })
 return}