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

// 自定义判定执行结果
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")
}

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])
  }
}
// 清理特定bucket
func (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
}

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.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
  }
  // 如果执行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
}