10 August 2019

本文学习参考自:Circuit Breaker patterncep21/circuit

业务问题场景

在业务系统中,通常存在服务之间的相互调用,例如服务A调用服务B,当出现如下情形:

  1. 服务A与服务B之间的网络出现异常
  2. 服务B过载
  3. 服务B出现异常

服务A应该减少对服务B的调用量,甚至服务A应该停止调用服务B,有必要的话,还可以采取相应的降级措施`。当服务B恢复正常后,才开始继续调用服务B。

断路器模式

在家庭电路中有一个叫断路器的安全设备,当出现电路过载、短路、漏电等情况时,就会发生跳闸,防止出现安全事故。类比到上面描述的业务问题场景,我们需要在系统中实现一个类似断路器功能的组件,用于阻止系统A重复尝试很可能失败的调用。

在断路器模式中,断路器组件需要监测到最近失败的调用,并且利用这些信息去决定新的调用是否执行,还是立即抛出异常。当断路器组件“跳闸”之后,还需要能探测被调用服务是否恢复正常,

断路器模式的代码实现,使用了有限状态机的思想。最基本的实现有三种状态:

  • 关闭(Closed):调用正常执行。断路器组件对最近失败的调用进行计数,当达到阈值时,则断路器组件“跳闸”,进入“打开”状态。
  • 打开(Open):调用请求会立即失败,断路器组件抛出异常。
  • 半打开(Half-Open):当处于“打开”状态时,会启动一个超时定时器,当超时后,断路器组件会进入“半打开”状态,此时允许执行部分调用,断路器会对成功执行的调用进行计数,达到阈值后,会认为被调用服务恢复正常,断路器状态回到“关闭”状态,如果有请求出现失败,则回到“打开”状态。

下图来自:Circuit Breaker pattern

问题和注意事项

  • 异常处理:系统需要考虑到断路器抛出的各类异常该如何处理。比如采取降级措施,把请求转发给备份服务,或者通知上游稍后重试。
  • 异常的类型:服务调用请求可能出现超时,或网络不通,下游服务明确返回失败的情况,断路器可能需要针对不同情况的错误,采取不同的状态切换策略。例如触发切换到“打开”状态的条件,可以是超时错误的阈值比下游服务明确返回失败的阈值更高。
  • 日志:断路器需要记录下所有失败的请求,方便相关人员监控定位问题。
  • 恢复:配置合适的策略,让断路器检测下游服务是否恢复正常,
  • “打开”到“半打开”的状态切换:可以不使用定时器,而是周期性的探测下游服务是否恢复。
  • 人为干预:服务异常恢复需要的时间有长有短,断路器最好能提供人为控制的接口,方便将断路器强制切换到“打开”或“关闭”状态。
  • 并发:一个断路器可能会被很多请求并发访问,所以断路器工程化实现所需的时间和空间消耗需要尽量的小。
  • 资源差异:为不同的资源访问,单独创建相应的断路器。
  • 加速“跳闸”:当可以从下游服务获取到足够明确的异常时,则立即切换到“打开”状态。

Golang中断路器的实现:cep21/circuit

github.com/cep21/circuit 实现了类似 Hystrix(Java版本)的断路器模式。在示例代码circuit/v3/example/main.go中,模拟了服务调用可能出现的各种情况,以及对应的断路器配置,以下我总结了对该库的理解和应用。

cep21/circuit中主要的类型和接口:

circuit.Manager
// 管理多个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
}
circuit.Circuit
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的解释看是为了方便测试,当没在测试代码里有看到使用
}
circuit.ClosedToOpen
type ClosedToOpen interface {
    RunMetrics  // 统计调用出现的各种情况
    Metrics     // 统计状态切换的情况
    // 当出现ErrFailure和ErrTimeout的失败调用时,会调用ShouldOpen,ShouldOpen会根据RunMetrics信息决定是否切换到“打开”状态
    ShouldOpen(now time.Time) bool
    // 即使断路器处于“关闭”状态,也希望能阻止调用
    Prevent(now time.Time) bool
}

circuit/v3/closers/hystrix/opener.go 是 hystrix 的默认实现,实现了 circuit.ClosedToOpen 接口。

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 是滑动窗口计数器,用于统计一段时间窗口内,对每个时间片上发生的事件进行计数
circuit.OpenToClosed
type OpenToClosed interface {
    RunMetrics  // 统计调用出现的各种情况
    Metrics     // 统计状态切换的情况
    // 当调用成功时,会调用ShouldClose,ShouldClose会根据RunMetrics信息决定是否切换到“关闭”状态
    ShouldClose(now time.Time) bool
    // “半打开”状态的实现,用于在断路器处于“打开”状态时,允许部分调用执行
    Allow(now time.Time) bool
}

circuit/v3/closers/hystrix/closer.go 是 hystrix 的默认实现,实现了 circuit.OpenToClosed 接口。

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
}

cep21/circuit 实现的断路器模式有限状态机图解

异常处理

func (c *Circuit) Execute(ctx context.Context, runFunc func(context.Context) error, fallbackFunc func(context.Context, error) error) error 方法返回的 error 可能的情况有:

  • runFunc 返回的 error,包括 circuit.BadRequest
  • circuit 返回的
    • 当状态为“打开”:circuit.circuitError{concurrencyLimitReached: true, msg: "circuit is open"}
    • 当并发请求超过阈值:
      • runFunc并发超过阈值:circuit.circuitError{concurrencyLimitReached: true, msg: "throttling connections to command"}
      • fallbackFunc并发超过阈值:circuit.circuitError{circuitOpen: true, msg: "throttling concurrency to fallbacks"}
  • fallbackFunc 返回的 error
  • nil

其中 fallbackFunc 的 error 参数可能为:

  • runFunc 返回的 error,除了 circuit.BadRequest
  • circuit 返回的
    • 当状态为“打开”:circuit.circuitError{concurrencyLimitReached: true, msg: "circuit is open"}
    • 当并发请求超过阈值:
      • runFunc并发超过阈值:circuit.circuitError{concurrencyLimitReached: true, msg: "throttling connections to command"}
  • nil

业务系统中的实际应用

在我的业务系统中,OpenToClosed接口的实现,不打算使用hystrix默认的进入“半打开”的逻辑:定时放行部分调用。因为这样可能会影响到上游的业务请求,并且在fallbackFunc中,我会去调用异地热备服务。所以进入“半打开”状态,我选择自己实现OpenToClosed接口,策略如下:

  • 当断路器进入“打开”状态时,启动下游服务健康检查定时器,通过模拟业务请求调用下游服务
    • 如果调用成功了,并且累积成功调用次数达到一定阈值,此时OpenToClosed.Allow(now time.Time) bool根据一定概率返回 true,断路器进入“半打开”状态
    • 如果调用失败了,OpenToClosed.Allow(now time.Time) bool返回 false
  • 当断路器进入“半打开”状态
    • 如果调用成功了,进行计数,达到阈值后,断路器进入“关闭”状态,并且停止下游健康检查定时器
    • 如果调用失败了,断路器回到“打开”状态

实现代码:

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