Golang微服务容错机制与降级策略

context

context.WithTimeout

context.WithDeadline

package main

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

func fetchWithTimeout(ctx context.Context, url string) (string, error) {
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return "", fmt.Errorf("创建请求失败: %w", err)
    }

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        // context.DeadlineExceeded error will be wrapped here if timeout occurs
        return "", fmt.Errorf("请求执行失败: %w", err)
    }
    defer resp.Body.Close()

    body, err := ioutil.ReadAll(resp.Body)
    if err != nil {
        return "", fmt.Errorf("读取响应体失败: %w", err)
    }
    return string(body), nil
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    data, err := fetchWithTimeout(ctx, "http://localhost:8080/slow_service") // 假设这是一个慢服务
    if err != nil {
        fmt.Printf("获取数据失败: %v\n", err)
        if ctx.Err() == context.DeadlineExceeded {
            fmt.Println("操作超时了！")
        }
        return
    }
    fmt.Printf("获取到数据: %s\n", data)
}

github.com/sethgrid/retry

package main

import (
    "fmt"
    "math/rand"
    "time"
)

func callExternalService() (string, error) {
    // 模拟外部服务调用，有一定几率失败
    if rand.Intn(10) < 7 { // 70% 失败率
        return "", fmt.Errorf("外部服务暂时不可用")
    }
    return "数据已成功获取", nil
}

func main() {
    maxRetries := 5
    baseDelay := 100 * time.Millisecond // 初始延迟
    var result string
    var err error

    for i := 0; i < maxRetries; i++ {
        fmt.Printf("尝试调用外部服务 (第 %d 次)\n", i+1)
        result, err = callExternalService()
        if err == nil {
            fmt.Printf("成功: %s\n", result)
            return
        }

        fmt.Printf("失败: %v\n", err)

        if i < maxRetries-1 {
            // 指数退避 + 抖动
            delay := baseDelay * time.Duration(1<<i)
            jitter := time.Duration(rand.Int63n(int64(delay / 2))) // 随机抖动，最大为当前延迟的一半
            sleepTime := delay + jitter
            fmt.Printf("等待 %v 后重试...\n", sleepTime)
            time.Sleep(sleepTime)
        }
    }
    fmt.Printf("多次重试后仍失败: %v\n", err)
}

github.com/sony/gobreaker

package main

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

    "github.com/sony/gobreaker"
)

var cb *gobreaker.CircuitBreaker

func init() {
    // 配置熔断器
    settings := gobreaker.Settings{
        Name:        "ExternalServiceBreaker",
        MaxRequests: 3,                 // 半开状态下允许通过的请求数
        Interval:    5 * time.Second,   // 统计周期
        Timeout:     10 * time.Second,  // 打开状态持续时间
        ReadyToTrip: func(counts gobreaker.Counts) bool {
            // 当请求总数大于等于3个，且失败率大于60%时，熔断器跳闸
            return counts.Requests >= 3 && float64(counts.Failure)/float64(counts.Requests) >= 0.6
        },
        OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
            fmt.Printf("熔断器 '%s' 状态从 %s 变为 %s\n", name, from, to)
        },
    }
    cb = gobreaker.NewCircuitBreaker(settings)
}

func callExternalServiceWithBreaker() (string, error) {
    // 使用熔断器执行操作
    result, err := cb.Execute(func() (interface{}, error) {
        // 模拟实际的外部服务调用
        resp, err := http.Get("http://localhost:8081/unstable_service") // 假设这是一个不稳定的服务
        if err != nil {
            return nil, err
        }
        defer resp.Body.Close()

        if resp.StatusCode != http.StatusOK {
            return nil, fmt.Errorf("服务返回非200状态码: %d", resp.StatusCode)
        }

        body, err := ioutil.ReadAll(resp.Body)
        if err != nil {
            return nil, err
        }
        return string(body), nil
    })

    if err != nil {
        return "", err
    }
    return result.(string), nil
}

func main() {
    for i := 0; i < 20; i++ {
        data, err := callExternalServiceWithBreaker()
        if err != nil {
            fmt.Printf("第 %d 次调用失败: %v\n", i+1, err)
            // 如果是熔断器打开导致的错误，可以进行降级处理
            if err == gobreaker.ErrOpenState {
                fmt.Println("熔断器已打开，执行快速失败或降级策略。")
            }
        } else {
            fmt.Printf("第 %d 次调用成功: %s\n", i+1, data)
        }
        time.Sleep(500 * time.Millisecond) // 模拟间隔调用
    }
}

package main

import (
    "fmt"
    "sync"
    "time"
)

// 定义一个信号量来限制并发访问
type Semaphore chan struct{}

func NewSemaphore(n int) Semaphore {
    return make(Semaphore, n)
}

func (s Semaphore) Acquire() {
    s <- struct{}{}
}

func (s Semaphore) Release() {
    <-s
}

func callServiceA(id int) {
    fmt.Printf("Service A: 请求 %d 开始\n", id)
    time.Sleep(time.Duration(2+id%3) * time.Second) // 模拟Service A可能很慢
    fmt.Printf("Service A: 请求 %d 结束\n", id)
}

func callServiceB(id int) {
    fmt.Printf("Service B: 请求 %d 开始\n", id)
    time.Sleep(500 * time.Millisecond) // 模拟Service B通常很快
    fmt.Printf("Service B: 请求 %d 结束\n", id)
}

func main() {
    // 为Service A分配一个较小的并发池（舱壁）
    serviceASemaphore := NewSemaphore(3) // 限制Service A最多3个并发请求

    // 为Service B分配一个较大的并发池
    serviceBSemaphore := NewSemaphore(10) // 限制Service B最多10个并发请求

    var wg sync.WaitGroup

    // 模拟对Service A的请求
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            serviceASemaphore.Acquire() // 获取Service A的资源
            defer serviceASemaphore.Release() // 释放Service A的资源
            callServiceA(id)
        }(i)
    }

    // 模拟对Service B的请求
    for i := 0; i < 20; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            serviceBSemaphore.Acquire() // 获取Service B的资源
            defer serviceBSemaphore.Release() // 释放Service B的资源
            callServiceB(id)
        }(i)
    }

    wg.Wait()
    fmt.Println("所有请求完成。")
}