Golang熔断器的开发过程详解

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

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

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  
}

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

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