Golang

关注公众号 jb51net

关闭
首页 > 脚本专栏 > Golang > Golang熔断器

Golang熔断器的开发过程详解

作者：真的不想写了

Golang熔断器是一种用于处理分布式系统中服务调用的故障保护机制,它可以防止故障服务的连锁反应,提高系统的稳定性和可靠性,本文将给大家详细的介绍一下Golang熔断器的开发过程,需要的朋友可以参考下

为什么需要熔断

在分布式大行其道的今天，子系统与子系统之间通常使用RPC进行通信，但由于是远程调用，如果整条链路中的某个服务出现异常，便可能雪崩从而导致整个系统崩溃（亲身经历）。

如何解决

重试

我们可以很容易的想到加入重试机制以及超时时间来解决：在指定时间内如果没有返回，便触发重试，达到重试阈值后如果还是没有拿到正确的数据变做降级处理。

这样做确实能够保证整个系统不会雪崩，单次请求也能拿到合理的数据。但因为重试机制，让本就摇摇欲坠的故障服务雪上加霜，最后可能导致该服务完全不可用。

熔断

重试行不通的原因是因为它还会继续调用故障服务，甚至请求量比平时还翻了几倍。所以我们的核心问题是解决出现故障后继续调用这个问题，此时就可以引入熔断器了。

简单来说，熔断器会收集每次远程调用的结果，并根据一定的规则判断目标服务是否出现故障，如果出现故障，就不再调用，直接进入降级处理并返回。

总体设计

err := hystrix.Do("test", func() error {  
    // do something
    return nil  
}, func(err error) error {  
    fmt.Println(err.Error())  
    return err  
})

实现

指标

type (  
    metrics struct {  
        total *number  
        success *number  
        fail *number  
    }  
    number struct {  
        buckets [10]*bucket  
        mutex sync.RWMutex  
    }  
    bucket struct {  
        timestamp int64  
        value float64  
    }  
)

我们定义了三个结构体，bucket用于存储上报量，number使用环形数组存储最近10秒的上报量，而metrics存储了请求总量、成功量以及失败量。

这三个结构体中，我们重点关注指标的上报（increment）以及读操作即可（sum）

func (number *number) increment(now int64) {  
    index := now % 10  
    number.mutex.RLock()  
    if now < number.buckets[index].timestamp {  
        number.mutex.RUnlock()  
        return  
    }  
    number.mutex.RUnlock()  
    number.mutex.Lock()  
    defer number.mutex.Unlock()  
    if number.buckets[index].timestamp != now {  
        number.buckets[index].value = 0  
    }  
    number.buckets[index].timestamp = now  
    number.buckets[index].value++  
}

首先我们需要拿到当前时间戳对应的环形数组下标，然后加锁，判断当前时间戳是否有效。注意第13～15行代码，如果该下标当前存储的数据是历史数据，那么重新赋值、覆盖就好了（这就是环形数组的优势，无需扩容以及执行delete操作）。

func (number *number) sum() (sum float64) {  
    number.mutex.RLock()  
    defer number.mutex.RUnlock()  
    now := time.Now().Unix()  
    for _, ele := range number.buckets {  
        if ele.timestamp <= now-10 {  
            continue  
        }  
        sum += ele.value  
    }  
    return  
}

如果环形数组中存储的指标数据是10秒之前的，那么就不参与计算。

熔断器

有了指标数据，我们就可以考虑如何通过它来判断目标服务是否出现故障了。最简单的，我们可以定义一个阈值：当最近10秒的请求错误率达到这个阈值后，就认为目标服务出现故障。

但这个判断得基于一定的请求量才能开启，否则得到的错误率与目标服务当前的运行状态对比会存在一定误差，例如服务启动后第一次请求目标服务，但因为一些偶现原因返回了error，那么此时的错误率就是100%，认为目标服务出现故障，后续请求都会被拦截。

熔断器开启后，还得想办法将它关闭，否则就算目标服务恢复了正常，熔断器还是会将该请求拦截。我们可以设置一个时间窗口并记录熔断器打开的时间，只要过了这个时间窗口，我们便可以关闭熔断器并重新收集指标数据进行再次进行判断。

type Circuit struct {  
    Timeout time.Duration  
    RequestVolumeThreshold int // 达到这个请求数量后才去判断是否要开启熔断  
    ErrorPercentThreshold int // 请求数量大于等于 RequestVolumeThreshold 并且错误率到达这个百分比后就会启动熔断  
    SleepWindow int // 熔断器被打开后 SleepWindow 的时间就是控制过多久后去尝试服务是否可用了 单位为毫秒  
    open bool  
    lastOpenTime int64 // 单位ms  
    mutex sync.RWMutex  
    metric *metrics  
}
func (circuit *Circuit) isHealthy() bool {  
    // 当前总请求量小于设置的阈值 返回
    if int(circuit.metric.totalRequest()) < circuit.RequestVolumeThreshold {  
        return true  
    }  
    // 判断错误率是否大于设定的阈值，从而判断目标服务是否出现故障
    return circuit.metric.errorPercent() < circuit.ErrorPercentThreshold  
}
func (circuit *Circuit) isOpen() bool {  
    circuit.mutex.RLock()  
    o := circuit.open  
    circuit.mutex.RUnlock()  
    if !o {  
        return false  
    }  
    // 当前时间与熔断器打开时间进行对比，如果过了时间窗口，那么恢复。
    if circuit.lastOpenTime+int64(circuit.SleepWindow) < time.Now().UnixMilli() {  
        circuit.setClose()  
        return false  
    }  
    return true  
}  
func (circuit *Circuit) setClose() {  
    circuit.mutex.Lock()  
    defer circuit.mutex.Unlock()  
    if !circuit.open {  
        return  
    }  
    circuit.open = false  
    // 清空指标数据 重新计算
    circuit.metric.clear()  
}

执行过程

首先我们需要判断熔断器是否是打开的状态，如果是，那么直接降级处理。如果不是，便执行传入的run()函数，得到返回结果并上报。

func (cmd *command) do() error {  
    defer cmd.reportAllEvents()  
    // 判断熔断器是否打开
    if !cmd.circuit.allowRequest() {  
        cmd.report(circuitOpenEvent)  
        return cmd.tryFallback(ErrCircuitOpen)  
    }  
    // 设置超时时间
    timer := time.NewTimer(cmd.circuit.Timeout)  
    defer timer.Stop()  
    finish, errCh := make(chan struct{}), make(chan error)  
    go func() {  
        if err := cmd.run(); err != nil {  
            errCh <- err  
            return  
        }  
        finish <- struct{}{}  
    }()  
    // 处理 超时、执行成功、执行失败 这三种情况
    // 超时以及执行失败都认为错误，降级处理
    select {  
    case <-timer.C:  
        return cmd.tryFallback(ErrTimeout)   
    case <-finish:  
        cmd.report(successEvent)  
        return nil  
    case err := <-errCh:  
        return cmd.tryFallback(err)  
    }  
}

以上就是熔断器的全部思路以及核心代码。我们通过metrics来收集指标并使用Circuit配置熔断规则以及根据metrics收集的指标判断目标服务是否出现故障，最后使用Command来执行配置的run()函数以及降级逻辑。

项目地址

https://gitee.com/colocust/hystrix

以上就是Golang熔断器的开发过程详解的详细内容，更多关于Golang熔断器的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:
阅读全文