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

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

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

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

// 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()
    // 記錄錯誤日志內(nèi)容
    ew.reasons[ew.index] = fmt.Sprintf("%s %s", time.Now().Format(timeFormat), reason)
    // 對 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")
}

// core/collection/rollingwindow.go

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

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 {
    // 桶，一個桶就是一個時間間隔
    buckets []*Bucket
    // 窗口大小，也就是桶的數(shù)量
    size    int
}

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)過了多少個時間間隔，也就是多少個桶
    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)過了多少個時間間隔，也就是多少個桶
    span := rw.span()
    // 還在同一單元時間內(nèi)不需要更新
    if span <= 0 {
        return
    }

    offset := rw.offset
    // reset expired buckets
    // 這里是清除過期桶的數(shù)據(jù)
    // 也是對數(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)該是一個時間的對齊，保持在桶內(nèi)指向位置是一致的
    rw.lastTime = now - (now-rw.lastTime)%rw.interval
}

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

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

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

// core/breaker/googlebreaker.go

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

// 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 {
    // 獲取最近一段時間的統(tǒng)計(jì)數(shù)據(jù)
    accepts, total := b.history()
    // 根據(jù)上文提到的算法來計(jì)算一個概率
    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
}

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

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

// req - 熔斷對象方法
// fallback - 自定義快速失敗函數(shù)，可對熔斷產(chǎn)生的err進(jìn)行包裝后返回
// acceptable - 對本次未熔斷時執(zhí)行請求的結(jié)果進(jìn)行自定義的判定，比如可以針對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
}

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

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