如何在Golang微服务中实现熔断机制 集成Hystrix-go与弹性模式设计

import (
    "fmt"
    "io/ioutil"
    "net/http"
    "time"

    "github.com/afex/hystrix-go/hystrix"
)

func init() {
    // 配置Hystrix命令
    hystrix.ConfigureCommand("my_downstream_service", hystrix.CommandConfig{
        Timeout:                1000,                           // 单次请求超时时间，单位毫秒
        MaxRequests:            10,                             // 熔断器打开前，在一个统计窗口内允许的最大请求数
        ErrorPercentThreshold:  25,                             // 错误率达到此阈值时，熔断器打开
        SleepWindow:            5000,                           // 熔断器打开后，等待多长时间进入半开状态（尝试请求）
        RequestVolumeThreshold: 5,                              // 在一个统计窗口内，请求数达到此阈值才开始计算错误率
    })
}

func callDownstreamService() (string, error) {
    output := ""
    err := hystrix.Do("my_downstream_service", func() error {
        // 这里是实际的业务逻辑，调用下游服务
        resp, err := http.Get("http://localhost:8081/data") // 假设这是你的下游服务地址
        if err != nil {
            return err
        }
        defer resp.Body.Close()
        body, err := ioutil.ReadAll(resp.Body)
        if err != nil {
            return err
        }
        output = string(body)
        return nil
    }, func(err error) error {
        // 回退逻辑，当主逻辑失败或熔断器打开时执行
        fmt.Printf("Fallback executed due to: %v\n", err)
        output = "Fallback data: Service temporarily unavailable."
        return nil // 返回nil表示回退成功，不返回错误
    })

    if err != nil {
        // 如果熔断器打开且回退也失败了，或者回退逻辑本身报错
        return "", fmt.Errorf("call downstream service failed: %w", err)
    }
    return output, nil
}

func main() {
    // 模拟多次调用
    for i := 0; i < 20; i++ {
        data, err := callDownstreamService()
        if err != nil {
            fmt.Printf("Attempt %d: Error: %v\n", i+1, err)
        } else {
            fmt.Printf("Attempt %d: Data: %s\n", i+1, data)
        }
        time.Sleep(500 * time.Millisecond) // 模拟请求间隔
    }
}

package main

import (
    "fmt"
    "io/ioutil"
    "net/http"
    "time"

    "github.com/afex/hystrix-go/hystrix"
    "github.com/afex/hystrix-go/hystrix/metricscollector"
    _ "github.com/afex/hystrix-go/hystrix/metricscollector/gorillastats" // 导入默认的度量收集器
)

// 模拟一个下游服务，它可能会失败或变慢
func mockDownstreamService(fail bool, slow bool) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        if slow {
            time.Sleep(1500 * time.Millisecond) // 模拟慢响应
        }
        if fail {
            w.WriteHeader(http.StatusInternalServerError)
            fmt.Fprintf(w, "Internal Server Error from mock service!")
            return
        }
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, "Hello from mock service!")
    }
}

func init() {
    // 配置Hystrix命令
    hystrix.ConfigureCommand("get_user_profile", hystrix.CommandConfig{
        Timeout:                500,  // 500ms超时
        MaxRequests:            5,    // 在熔断器打开前，在一个统计窗口内允许的最大请求数
        ErrorPercentThreshold:  20,   // 错误率达到20%时，熔断器打开
        SleepWindow:            3000, // 熔断器打开后，等待3秒进入半开状态
        RequestVolumeThreshold: 3,    // 在一个统计窗口内，请求数达到3个才开始计算错误率
    })

    // 可以为不同的命令组配置不同的线程池
    // hystrix.ConfigureCommand("get_user_profile", hystrix.CommandConfig{
    //  MaxConcurrentRequests: 10, // 限制并发请求数，实现舱壁隔离
    // })
}

func getUserProfile() (string, error) {
    var result string
    err := hystrix.Do("get_user_profile", func() error {
        // 模拟调用下游服务
        resp, err := http.Get("http://localhost:8081/user") // 假设下游服务地址
        if err != nil {
            return err
        }
        defer resp.Body.Close()

        if resp.StatusCode != http.StatusOK {
            return fmt.Errorf("downstream service returned non-OK status: %d", resp.StatusCode)
        }

        body, err := ioutil.ReadAll(resp.Body)
        if err != nil {
            return err
        }
        result = string(body)
        return nil
    }, func(err error) error {
        // 回退逻辑
        fmt.Printf("[Fallback] Executed for get_user_profile due to: %v\n", err)
        result = "Fallback User Profile: Default User" // 提供默认数据
        return nil // 返回nil表示回退成功
    })

    if err != nil {
        return "", fmt.Errorf("failed to get user profile: %w", err)
    }
    return result, nil
}

func main() {
    // 启动一个模拟的下游服务
    go func() {
        http.HandleFunc("/user", mockDownstreamService(false, false)) // 默认不失败，不慢
        // 在测试熔断时，可以改为：
        // http.HandleFunc("/user", mockDownstreamService(true, false)) // 模拟失败
        // http.HandleFunc("/user", mockDownstreamService(false, true)) // 模拟慢响应
        fmt.Println("Mock downstream service listening on :8081")
        http.ListenAndServe(":8081", nil)
    }()
    time.Sleep(100 * time.Millisecond) // 等待下游服务启动

    fmt.Println("--- Testing Hystrix Circuit Breaker ---")

    // 模拟请求，观察熔断行为
    for i := 0; i < 20; i++ {
        // 在某个时间点可以切换下游服务的行为来观察熔断
        // 例如，在第5次请求后，让下游服务开始失败
        if i == 5 {
            fmt.Println("\n--- Making downstream service fail/slow for testing ---")
            // 重新启动模拟服务，使其失败或变慢
            go func() {
                http.ListenAndServe(":8081", mockDownstreamService(true, true)) // 既失败又慢
            }()
            time.Sleep(100 * time.Millisecond)
        }

        profile, err := getUserProfile()
        if err != nil {
            fmt.Printf("Attempt %d: Error: %v\n", i+1, err)
        } else {
            fmt.Printf("Attempt %d: Profile: %s\n", i+1, profile)
        }
        time.Sleep(300 * time.Millisecond) // 模拟请求间隔
    }

    // 观察熔断器状态，Hystrix提供了一个流，可以通过Netflix Turbine / Hystrix Dashboard来可视化
    // 这里只是简单等待，实际应用中会集成监控系统
    time.Sleep(5 * time.Second)
    fmt.Println("\n--- End of Test ---")
}