Golang中使用断路器

// 管理多个circuits对象实例
type Manager struct {
    // func (h *Manager) CreateCircuit(name string, configs ...Config) (*Circuit, error) 方法创建circuits对象实例时，使用的配置，会按照逆序将多个配置合并为最终的配置
    DefaultCircuitProperties []CommandPropertiesConstructor

    // 每个circuits会有一个唯一命名的标识
    circuitMap map[string]*Circuit
    // 用于circuitMap的读写锁
    mu sync.RWMutex
}

type Circuit struct {
    //circuitStats
    CmdMetricCollector      RunMetricsCollection        // 统计调用出现的各种情况
    FallbackMetricCollector FallbackMetricsCollection   // 统计降级调用出现的各种情况
    CircuitMetricsCollector MetricsCollection           // 统计Circuit状态切换的情况
    // This is used to help run `Go` calls in the background
    goroutineWrapper goroutineWrapper // 用于异步调用的封装
    name             string           // 断路器唯一命名的标识
    notThreadSafeConfig Config // 非线程安全的断路器配置
    notThreadSafeConfigMu sync.Mutex
    threadSafeConfig      atomicCircuitConfig // 线程安全的断路器配置

    // Tracks if the circuit has been shut open or closed
    isOpen faststats.AtomicBoolean // 断路器只有“打开”和“关闭”两种状态

    // Tracks how many commands are currently running
    concurrentCommands faststats.AtomicInt64            // 统计有多少并发调用
    // Tracks how many fallbacks are currently running
    concurrentFallbacks faststats.AtomicInt64           // 统计有多少降级的并发调用

    // ClosedToOpen controls when to open a closed circuit
    ClosedToOpen ClosedToOpen   // 控制断路器由“关闭”状态切换到“打开”状态
    // openToClosed controls when to close an open circuit
    OpenToClose OpenToClosed    // 控制断路器由“打开”状态切换到“关闭”状态

    timeNow func() time.Time    // 对time.Now的封装，值始终为config.General.TimeKeeper.Now，从config.TimeKeeper的解释看是为了方便测试，当没在测试代码里有看到使用
}

type ClosedToOpen interface {
    RunMetrics  // 统计调用出现的各种情况
    Metrics     // 统计状态切换的情况
    // 当出现ErrFailure和ErrTimeout的失败调用时，会调用ShouldOpen，ShouldOpen会根据RunMetrics信息决定是否切换到“打开”状态
    ShouldOpen(now time.Time) bool
    // 即使断路器处于“关闭”状态，也希望能阻止调用
    Prevent(now time.Time) bool
}

type Opener struct {
    errorsCount             faststats.RollingCounter // 统计调用出现的ErrFailure和ErrTimeout的情况
    legitimateAttemptsCount faststats.RollingCounter // 统计调用出现的ErrFailure和ErrTimeout，以及Success的情况

    errorPercentage        faststats.AtomicInt64 // 错误阈值
    requestVolumeThreshold faststats.AtomicInt64 // 如果在一段时间窗口内的调用次数小于该阈值，则不会将断路器切换到“打开”状态

    mu     sync.Mutex // 修改config时的互斥锁
    config ConfigureOpener
}

// faststats.RollingCounter 是滑动窗口计数器，用于统计一段时间窗口内，对每个时间片上发生的事件进行计数

type OpenToClosed interface {
    RunMetrics  // 统计调用出现的各种情况
    Metrics     // 统计状态切换的情况
    // 当调用成功时，会调用ShouldClose，ShouldClose会根据RunMetrics信息决定是否切换到“关闭”状态
    ShouldClose(now time.Time) bool
    // “半打开”状态的实现，用于在断路器处于“打开”状态时，允许部分调用执行
    Allow(now time.Time) bool
}

type Closer struct {
    // 当断路器处于“打开”状态时，定时放行部分调用，被OpenToClosed.Allow调用
    reopenCircuitCheck faststats.TimedCheck

    concurrentSuccessfulAttempts faststats.AtomicInt64 // 当调用为Success时，加1，当调用为ErrFailure、ErrTimeout时，重置为0
    closeOnCurrentCount          faststats.AtomicInt64 // 切换到“关闭”状态的阈值，在OpenToClosed.ShouldClose中使用

    mu     sync.Mutex // 修改config时的互斥锁
    config ConfigureCloser
}

package circuit

import (
    "math/rand"
    "sync"
    "time"

    "github.com/cep21/circuit"
    "github.com/cep21/circuit/v3/faststats"
)

type ConfigureCloser struct {
    CloseOnSuccessfulAttemptsCount int64
    ReopenHealthCheck              *HealthCheck
}

type Closer struct {
    reopenHealthCheck *HealthCheck

    successfulAttempts             faststats.AtomicInt64
    closeOnSuccessfulAttemptsCount int64
}

func NewCloser(config ConfigureCloser) circuit.OpenToClosed {
    return &Closer{
        reopenHealthCheck:              config.ReopenHealthCheck,
        closeOnSuccessfulAttemptsCount: config.CloseOnSuccessfulAttemptsCount,
    }
}

// start health check when circuit is opened
func (c *Closer) Opened(now time.Time) {
    c.reopenHealthCheck.start()
}

// stop health check when circuit is closed
func (c *Closer) Closed(now time.Time) {
    c.reopenHealthCheck.stop()
}

// half-open
func (c *Closer) Allow(now time.Time) bool {
    return c.reopenHealthCheck.ok()
}

func (c *Closer) Success(now time.Time, duration time.Duration) {
    c.successfulAttempts.Add(1)
}

func (c *Closer) ErrBadRequest(now time.Time, duration time.Duration) {
}

func (c *Closer) ErrInterrupt(now time.Time, duration time.Duration) {
}

func (c *Closer) ErrConcurrencyLimitReject(now time.Time) {
}

func (c *Closer) ErrShortCircuit(now time.Time) {
}

func (c *Closer) ErrFailure(now time.Time, duration time.Duration) {
    c.successfulAttempts.Set(0)
    c.reopenHealthCheck.reset()
}

func (c *Closer) ErrTimeout(now time.Time, duration time.Duration) {
    c.successfulAttempts.Set(0)
    c.reopenHealthCheck.reset()
}

func (c *Closer) ShouldClose(now time.Time) bool {
    return c.successfulAttempts.Get() > c.closeOnSuccessfulAttemptsCount
}

type ConfigureHealthCheck struct {
    TickDuration     time.Duration
    Run              func() bool
    Threshold        int64
    AllowProbability int
}

type HealthCheck struct {
    running      bool
    stopSignalCh chan struct{}
    count        faststats.AtomicInt64
    mu           sync.Mutex
    config       ConfigureHealthCheck
}

func NewHealthCheck(config ConfigureHealthCheck) *HealthCheck {
    return &HealthCheck{
        stopSignalCh: make(chan struct{}),
        config:       config,
    }
}

func (h *HealthCheck) start() {
    h.mu.Lock()
    defer h.mu.Unlock()

    if h.running {
        return
    }

    h.running = true
    h.count.Set(0)

    go func() {
        tick := time.Tick(h.config.TickDuration)
        for {
            select {
            case <-tick:
                if h.config.Run() {
                    h.count.Add(1)
                } else {
                    h.count.Set(0)
                }
            case <-h.stopSignalCh:
                return
            }
        }
    }()
}

func (h *HealthCheck) stop() {
    h.mu.Lock()
    defer h.mu.Unlock()

    if !h.running {
        return
    }

    h.stopSignalCh <- struct{}{}
    h.running = false
    h.count.Set(0)
}

func (h *HealthCheck) ok() bool {
    h.mu.Lock()
    defer h.mu.Unlock()

    if !h.running {
        return true
    }

    return h.count.Get() > h.config.Threshold && h.config.AllowProbability > rand.Intn(101)
}

func (h *HealthCheck) reset() {
    h.count.Set(0)
}